"البساطة شرط أساسي للاعتمادية" بقلم Edsger Dijkstraمقدمة
الرسوم البيانية هي بنية بيانات واضحة وسهلة الفهم ، منذ زمن ليونارد أويلر أجبرت على كسر عقول البشرية في مهام متنوعة ، مثل كيف يمكنك المرور عبر جميع الجسور السبعة في كونيجسبيرج دون المرور بأي منها مرتين أو كوسيط متنقل طريق مربح.
لقد تغير الكثير منذ أويلر: ظهرت الترانزستورات ولغات البرمجة والحوسبة الموزعة. هذه هي الأخيرة من هذه القائمة التي تبسط بشكل كبير تخزين ومعالجة الرسوم البيانية. في الواقع ، هذا ما سيتم مناقشته في هذه المقالة.
إذا لم تكن على دراية بمفاهيم Apache Spark الأساسية مثل RDD ، برنامج التشغيل ، عقدة العمال ، وما إلى ذلك ، ثم قبل متابعة هذه المقالة ، أوصيك بقراءة
الوثائق من Databricks.
بالنسبة لي ، فإن أفضل طريقة للتعامل مع التكنولوجيا هي محاولة كتابة شيء عليها. في هذه المقالة سنقوم بتحليل تشابه "الشبكة الاجتماعية" باستخدام المفاهيم الأساسية لنظرية الرسم البياني.
كتابة رمز
طريقة تخزين "شبكتنا الاجتماعية" التي اخترتها كانت بسيطة وبديهية للغاية: ملفات tsv على القرص ، بالطبع يمكن أن تكون ملفات بأي تنسيق آخر مثل Parquet ، Avro. لا يهم موقع تخزين الملفات إذا كان يمكن أن يكون HDFS أو S3 ، حتى إذا كنا بحاجة إلى تغيير شيء ما ، فإن Spark SQL ستقوم بالعمل الرئيسي بالنسبة لنا. ستكون بنية الشبكة كما يلي: الملف الأول هو زوج من معرف المستخدم واسمه ، وملف معرف المستخدم الثاني وقائمة من أقرانه. يدعم Apache Spark لغات برمجة Java و Scala و Python التالية كواجهات برمجة تطبيقات. اخترت الثانية.
على الفور أريد أن أجيب على السؤال الشائع عما إذا كان الأمر يستحق استخدام Spark GraphX لتخزين الرسوم البيانية عندما يكون لديك العديد من عمليات الإدراج / التحديث - الإجابة هي لا ، جميع عمليات تغيير RDD تفرض تغيير RDD بالكامل في المجموعة ، وهو ليس الحل الأمثل ، خاصة هي مناسبة لهذه الحالة حل NoSql مثل Neo4J أو Titanium أو حتى Cassandra ، Hbase. لا شيء يمنعك من استخدام Spark GraphX معهم خصيصًا لمعالجة الرسوم البيانية ، وتحميل البيانات نفسها من قاعدة البيانات ، على سبيل المثال ، بواسطة sheduler أو بأسلوب يحركه الحدث.
حسنًا ، فلنبدأ في كتابة الرمز. نحتاج أولاً إلى تحميل الرسم البياني في الذاكرة ، وأخذ الملفات المصدر وسحب الذروات والحواف الضرورية (تظهر النقاط الرئيسية هنا ، يمكنك العثور على الرابط إلى القائمة الكاملة مع شفرة المصدر في نهاية المقالة):
def verts: RDD[(VertexId, String)] = sc.textFile(USER_NAMES) .flatMap(InputDataFlow.parseNames) def edges: RDD[Edge[PartitionID]] = sc.textFile(USER_GRAPH) .flatMap(InputDataFlow.makeEdges)
بريجل
الآلية الرئيسية لتكرار الرسم البياني في GraphX هي خوارزمية Pregel. تم تطوير الخوارزمية بواسطة Google ، ويستخدم نموذج Pregel
نقل الرسائل بين القمم في الرسم البياني. الرسالة التي تمر عبر سلسلة من التكرارات تسمى supersteps هي الفكرة الرئيسية وراء هذه الخوارزمية. أيضا ، يمكن وصف الفكرة الرئيسية على النحو التالي:
"فكر مثل قمة الرأس" ، أي أن حالة القمة الحالية تعتمد فقط على حالة جيرانها.
يصبح Pregel ضروريًا للغاية عندما يصبح حل مشكلة باستخدام MapReduce العادي عملية صعبة للغاية. من المثير للاهتمام أن اسم Pregel يأتي من اسم النهر ، الذي امتد على جسور Koenigsberg السبعة.
البدائي الرئيسي لاجتياز رسم بياني هو ثلاثي - يتكون من المكونات التالية: القمة الحالية (قمة المصدر) ، القمة التي نمرر إليها (قمة الوجهة) والحافة بينهما (حافة توصيل) - كل شيء واضح: أين نذهب حيث نذهب وبأي طريق نذهب. أيضًا ، بالنسبة إلى Pregel ، يجب عليك تحديد المسافة الافتراضية بين القمم ، كقاعدة ، إنها وظيفة PositiveInfinity ، و UDF (وظيفة يحددها المستخدم) لكل رأس لمعالجة الرسالة الواردة وحساب القمة التالية ، و UDF لدمج الرسالتين الواردتين ، يجب أن تكون هذه الوظيفة تبادلية و النقابي. نظرًا لأن Scala هي لغة وظيفية ، فسيتم تمثيل الدالتين الأخيرتين على أنهما تعبيران لامدا.
عندما نفكك المكونات الرئيسية لبريجل ، من المفيد أن نتدرب. الخوارزمية الأولى التي سننفذها هي خوارزمية Dijkstra للعثور على أقصر مسار من قمة اعتباطية إلى جميع الآخرين.
def dijkstraShortestPath[VT](graph: GenericGraph[VT], sourceId: VertexId) = { val initialGraph = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity) val sssp = initialGraph.pregel(Double.PositiveInfinity)( (_, dist, newDist) => math.min(dist, newDist), triplet => {
كل شيء واضح هنا: نبدأ من قمة معينة ، ونستخدم الحد الأدنى من الوظيفة لتحديد الحد الأدنى للمسافة في كل خطوة. تحافظ الوظيفة الأولى المستخدمة من قبل Pregel على أقصر مسافة بين الرسالة الواردة والرأس الحالي. الوظيفة الثانية توزع الرسائل على الجيران مع الحفاظ على المسافة. الوظيفة الأخيرة - هذه تمثيلية لمرحلة التخفيض - تحدد القيمة الدنيا في حالة وجود العديد من الرسائل الواردة. بعد ذلك ، نقوم ببساطة بتشكيل إخراج رسم بياني مناسب.
درجة الانفصال
أنا متأكد من أن العديد من قراء هذه المقالة قد سمعوا عن نظرية ستة مصافحات (
ست درجات من الفصل ) - هذه نظرية غير مثبتة وفقًا لها يفصل بين أي شخصين ما لا يزيد عن خمسة مستويات من المعارف الشائعة ، أي أنه يلزم الحد الأقصى من 6 مصافحات من أجل ربط اثنين من المصافحة العشوائية رجل على وجه الأرض. من حيث نظرية الرسم البياني ، يبدو مثل هذا: لا يتجاوز قطر الرسم البياني للتاريخ 6 لأي شخصين على الأرض.
لنبدأ في كتابة الكود مع ما يلي ، نحتاج إلى بحث أول على الرسم البياني للبحث عن جهات الاتصال للرأس المشار إليه ، لذلك نحتاج إلى تعديل كود خوارزمية Dijkstra:
def getBFS(root: VertexId) = { val initialGraph = graph.mapVertices((id, _) => if (id == root) 0.0 else Double.PositiveInfinity) val bfs = initialGraph.pregel(Double.PositiveInfinity, maxIterations = 10)( (_, attr, msg) => math.min(attr, msg), triplet => { if (triplet.srcAttr != Double.PositiveInfinity) { Iterator((triplet.dstId, triplet.srcAttr + 1)) } else { Iterator.empty } }, (a, b) => math.min(a, b)).cache() bfs }
كل شيء مشابه جدًا لما كان أعلاه ، لكننا نشير بالفعل إلى عدد التكرارات - قد يكون هذا رقمًا مختلفًا للرسم البياني - 10 للرسم البياني الذي حصلت عليه تجريبيًا. بعد ذلك ، انضم إلى أسماء المستخدمين وخذ أول 100 قيمة للمستخدم العشوائي:
def degreeOfSeparation(root: VertexId): Array[(VertexId, DegreeOfSeparation)] = { getBFS(root).vertices.join(verts).take(100) }
الآن نحن نبحث عن درجة الانفصال عن القمة المعطاة لجميع الآخرين ، يمكنك أيضًا البحث عن درجة الانفصال لذروتين تعسفيين:
def degreeOfSeparationTwoUser(firstUser: VertexId, secondUser: VertexId) = { getBFS(firstUser) .vertices .filter { case (vertexId, _) => vertexId == secondUser } .collect.map { case (_, degree) => degree } }
يمنحك Spark GraphX من الصندوق الفرصة للحصول على الكثير من المعلومات حول الرسم البياني ، على سبيل المثال ، للحصول على المكون المتصل من الرسم البياني (المكون المتصل):
def getMostConnectedUsers(amount: Int): Array[(VertexId, ConnectedUser)] = { graph.degrees.join(verts) .sortBy({ case (_, (userName, _)) => userName }, ascending = false) .take(amount) }
أو احصل على مقياس مثل عدد المثلثات في الرسم البياني (عدد المثلث):
def socialGraphTriangleCount = graph.triangleCount()
رتبة الصفحة
جاءت خوارزمية PageRank بفضل طلاب الدراسات العليا في ستانفورد لاري بيدج وسيرجي برين. لكل قمة من الرسم البياني ، تعين الخوارزمية أهمية بين جميع الرموز الأخرى. على سبيل المثال ، إذا كان لدى مستخدم Twitter عدد كبير من الاشتراكات من مستخدمين آخرين ، فسيحصل على تصنيف عالٍ ، وبالتالي ، يمكن العثور عليه بسهولة في محرك البحث.
يحتوي GraphX على إصدار ثابت وديناميكي لتطبيق PageRank. يحتوي الإصدار الثابت على عدد ثابت من التكرارات ، بينما ستعمل النسخة الديناميكية حتى يبدأ التقارب في التصنيف إلى القيمة المحددة.
بالنسبة إلى الرسم البياني لدينا ، سيكون هذا على النحو التالي:
def dynamicRanks(socialGraph: SocialGraph, tolerance: Double) = socialGraph.graph.pageRank(tol = tolerance).vertices def staticRanks(socialGraph: SocialGraph, tolerance: Double) = socialGraph.graph.staticPageRank(numIter = 20).vertices
الخلاصة
لاحظ قارئ يقظ أن موضوع هذه المقالة هو توزيع الرسوم البيانية ، ولكن عند كتابة التعليمات البرمجية ، لم نفعل شيئًا لجعل المعالجة موزعة حقًا. وهنا يجب أن نتذكر الاقتباس الذي قدمه Edsger Dijkstra في البداية. يبسط Spark حياتنا بشكل كبير من خلال تحمل عبء وأعباء الحوسبة الموزعة. كتابة التعليمات البرمجية التي سيتم تنفيذها على كتلة موزعة ليست مهمة صعبة كما قد يبدو في البداية. وهنا توجد أيضًا
العديد من الخيارات لإدارة موارد المجموعة: Hadoop YARN و Apache Mesos (شخصيًا ، خياري المفضل) ، ومؤخرًا ، هناك دعم لـ Kubernetes. يمكن العثور على جميع التعليمات البرمجية المصدر التي تم تحليلها في هذه المقالة على
جيثب .