स्पार्क ग्राफएक्स के साथ वितरित ग्राफ प्रोसेसिंग

"सरलता विश्वसनीयता के लिए पूर्वापेक्षा है" एद्गर दिज्क्स्ट्रा द्वारा

प्रस्तावना

रेखांकन डेटा संरचना को समझने के लिए इतना स्पष्ट और आसान है, लियोनहार्ड यूलर के समय से इसने मानव जाति के दिमाग को विविध कार्यों से तोड़ने के लिए मजबूर किया है, जैसे कि आप कोनिग्सबर्ग के सभी सात पुलों के माध्यम से दो बार या किसी यात्रा मध्यस्थ के बिना कैसे जा सकते हैं? लाभदायक मार्ग।

यूलर के बाद से बहुत कुछ बदल गया है: ट्रांजिस्टर, प्रोग्रामिंग भाषा और वितरित कंप्यूटिंग दिखाई दिए हैं। यह इस सूची का अंतिम है जिसने ग्राफ़ के भंडारण और प्रसंस्करण को नाटकीय रूप से सरल बना दिया है। दरअसल, इस लेख में चर्चा की जाएगी।

यदि आप मूल Apache Spark अवधारणाओं जैसे RDD, Driver program, Worker नोड आदि से परिचित नहीं हैं, तो इस लेख को जारी रखने से पहले, मैं आपको सलाह दूंगा कि आप Databricks से प्रलेखन पढ़ें।

मेरे लिए, एक तकनीक से निपटने का सबसे अच्छा तरीका यह है कि इस पर कुछ लिखने की कोशिश की जाए। इस लेख में हम ग्राफ सिद्धांत की मूल अवधारणाओं का उपयोग करके एक "सामाजिक नेटवर्क" की समानता का विश्लेषण करेंगे।

एक कोड लिखना

हमारे द्वारा चुने गए "सोशल नेटवर्क" को संग्रहीत करने की विधि अत्यंत सरल और सहज थी: डिस्क पर tsv फाइलें, स्वाभाविक रूप से ये किसी भी अन्य प्रारूप जैसे Parquet, Avro की फाइलें हो सकती हैं। फ़ाइलों का संग्रहण स्थान महत्वपूर्ण नहीं है अगर यह HDFS या S3 हो सकता है, भले ही हमें कुछ बदलने की आवश्यकता हो, तो स्पार्क एसक्यूएल हमारे लिए मुख्य काम करेगा। नेटवर्क संरचना निम्नानुसार होगी: पहली फ़ाइल उपयोगकर्ता आईडी और उसके नाम की जोड़ी, दूसरी उपयोगकर्ता आईडी फ़ाइल और साथियों की एक सूची है। अपाचे स्पार्क एपीआई के रूप में निम्नलिखित जावा, स्काला और पायथन प्रोग्रामिंग भाषाओं का समर्थन करता है। मैंने दूसरा चुना।

तुरंत मैं इस लोकप्रिय सवाल का जवाब देना चाहता हूं कि क्या यह ग्राफ़ को स्टोर करने के लिए स्पार्क ग्राफक्स का उपयोग करने के लायक है जब आपके पास कई इंसर्ट / अपडेट ऑपरेशंस होते हैं - इसका जवाब नहीं है, सभी आरडीडी चेंज ऑपरेशंस फोर्स को क्लस्टर में पूरे आरडीडी को बदलते हैं, जो कि इष्टतम समाधान नहीं है, विशेष इस मामले के लिए उपयुक्त हैं। NoSql सॉल्यूशन जैसे Neo4J, टाइटेनियम या यहां तक ​​कि Cassandra, Hbase। कुछ भी नहीं आप विशेष रूप से रेखांकन प्रसंस्करण के लिए उनके साथ स्पार्क ग्राफ एक्स का उपयोग करने से रोकते हैं, डेटा को डेटाबेस से स्वयं लोड करना, उदाहरण के लिए, शेड्यूलर या एक इवेंट संचालित शैली में।

तो ठीक है, चलो कोड लिखना शुरू करें। सबसे पहले, हमें ग्राफ़ को मेमोरी में लोड करने की ज़रूरत है, स्रोत फ़ाइलों को लें और आवश्यक कोने और किनारों को बाहर निकालें (यहां मुख्य बिंदु हैं, स्रोत कोड के साथ पूर्ण लिस्टिंग का लिंक लेख के अंत में पाया जा सकता है):

def verts: RDD[(VertexId, String)] = sc.textFile(USER_NAMES) .flatMap(InputDataFlow.parseNames) def edges: RDD[Edge[PartitionID]] = sc.textFile(USER_GRAPH) .flatMap(InputDataFlow.makeEdges) 

Pregel

ग्राफएक्स में ग्राफ पुनरावृत्ति के लिए मुख्य तंत्र प्रागेल एल्गोरिदम है। एल्गोरिथ्म Google द्वारा विकसित किया गया था, प्रागेल मॉडल ग्राफ में कोने के बीच संदेशों के हस्तांतरण का उपयोग करता है। सुपरस्टेप्स नामक पुनरावृत्तियों की एक श्रृंखला से गुजरने वाला संदेश इस एल्गोरिथम के पीछे मुख्य विचार है। इसके अलावा, मुख्य विचार को निम्नानुसार वर्णित किया जा सकता है: "एक शिखर की तरह सोचें" , अर्थात वर्तमान शीर्ष की स्थिति केवल अपने पड़ोसियों की स्थिति पर निर्भर करती है।

नियमित MapReduce के साथ एक समस्या को हल करते समय Pregel बेहद आवश्यक हो जाता है। दिलचस्प बात यह है कि प्रागेल नाम नदी के नाम से आया है, जो कोएनिग्सबर्ग के सात पुलों का विस्तार करता है।

ग्राफ ट्रैवर्सल के लिए मुख्य आदिम एक ट्रिपल है - इसमें निम्नलिखित घटक होते हैं: वर्तमान शीर्ष (एक स्रोत शीर्ष), जिस शीर्ष में हम गुजरते हैं (एक गंतव्य शीर्ष) और उनके बीच का किनारा (एक जोड़ने वाला किनारा) - सब कुछ स्पष्ट है: जहां हम जहां जाते हैं और जिस रास्ते से जाते हैं। इसके अलावा, प्रीगेल के लिए, आपको आवृत्तियों के बीच डिफ़ॉल्ट दूरी को निर्दिष्ट करना होगा, एक नियम के रूप में, यह आने वाले संदेश को संसाधित करने के लिए एक पॉजिटिव इनफिनिटी, यूडीएफ (उपयोगकर्ता परिभाषित फ़ंक्शन) फ़ंक्शन है और आने वाले संदेशों की गणना करने के लिए, और यूएफएफ दो आने वाले संदेशों को मर्ज करने के लिए, यह फ़ंक्शन कम्यूटेटिव होना चाहिए साहचर्य। चूंकि स्काला एक कार्यात्मक भाषा है, इसलिए अंतिम दो कार्यों को दो लंबोदर अभिव्यक्तियों के रूप में दर्शाया जाएगा।

जब हमने पहले से ही प्रागेल के मुख्य घटकों को अलग कर लिया है, तो अभ्यास के लिए आना सार्थक है। पहला एल्गोरिथ्म जिसे हम कार्यान्वित करेंगे, वह है दिक्जस्ट्रा का एल्गोरिदम, जो अन्य सभी के लिए एक मनमाना शीर्ष से सबसे छोटा रास्ता खोजने के लिए है।

 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 => { //Distance accumulator if (triplet.srcAttr + triplet.attr < triplet.dstAttr) { Iterator((triplet.dstId, triplet.srcAttr + triplet.attr)) } else { Iterator.empty } }, (a, b) => math.min(a, b) ) sssp.vertices.sortByKey(ascending = true).collect.mkString("\n") } 

यहां सब कुछ स्पष्ट है: हम एक दिए गए शीर्ष से शुरू करते हैं, प्रत्येक चरण पर न्यूनतम दूरी निर्धारित करने के लिए न्यूनतम फ़ंक्शन का उपयोग करते हैं। प्रागेल द्वारा उपयोग किया जाने वाला पहला फ़ंक्शन आने वाले संदेश और वर्तमान शीर्ष के बीच की सबसे छोटी दूरी को संरक्षित करता है। दूसरा फ़ंक्शन दूरी बनाए रखते हुए पड़ोसियों को संदेश वितरित करता है। अंतिम फ़ंक्शन - यह कम चरण का एक एनालॉग है - कई आने वाले संदेशों के मामले में न्यूनतम मूल्य का चयन करता है। अगला, हम बस एक सुविधाजनक ग्राफ आउटपुट बनाते हैं।

जुदाई की डिग्री

मुझे यकीन है कि इस लेख के कई पाठकों ने छह हैंडशेक ( पृथक्करण के छह डिग्री ) के सिद्धांत के बारे में सुना है - यह एक अप्रमाणित सिद्धांत है जिसके अनुसार किसी भी दो लोगों को पांच से अधिक सामान्य परिचितों द्वारा अलग नहीं किया जाता है, यानी दो मनमानी को जोड़ने के लिए अधिकतम 6 हैंडशेक की आवश्यकता होती है। पृथ्वी पर मनुष्य। ग्राफ सिद्धांत के संदर्भ में, यह इस तरह से लगता है: डेटिंग ग्राफ का व्यास पृथ्वी पर किसी भी दो लोगों के लिए 6 से अधिक नहीं है।

आइए निम्नलिखित के साथ कोड लिखना शुरू करें, हमें संकेतित शीर्ष के संपर्कों की खोज के लिए ग्राफ़ पर एक चौड़ाई-पहली खोज की आवश्यकता है, इसके लिए हमें दिक्जस्ट्रा एल्गोरिथ्म कोड को संशोधित करने की आवश्यकता है:

 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 } } 

बॉक्स से स्पार्क ग्राफएक्स आपको ग्राफ के बारे में बहुत सारी जानकारी प्राप्त करने का अवसर देता है, उदाहरण के लिए, ग्राफ के जुड़े घटक (जुड़ा हुआ घटक) प्राप्त करने के लिए:

 def getMostConnectedUsers(amount: Int): Array[(VertexId, ConnectedUser)] = { graph.degrees.join(verts) .sortBy({ case (_, (userName, _)) => userName }, ascending = false) .take(amount) } 

या ग्राफ़ में त्रिभुज की संख्या जैसे एक मीट्रिक प्राप्त करें (त्रिकोण गणना):

 def socialGraphTriangleCount = graph.triangleCount() 

पेज रैंक

पेजरैंक एल्गोरिथ्म स्टैनफोर्ड स्नातक छात्रों लैरी पेज और सर्गेई ब्रिन के लिए धन्यवाद के बारे में आया। ग्राफ़ के प्रत्येक शीर्ष के लिए, एल्गोरिथ्म अन्य सभी के बीच महत्व प्रदान करता है। उदाहरण के लिए, यदि एक ट्विटर उपयोगकर्ता के पास अन्य उपयोगकर्ताओं से बड़ी संख्या में सदस्यता है, तो उसके पास उच्च रेटिंग होगी, इसलिए, वह आसानी से खोज इंजन में पाया जा सकता है।

ग्राफएक्स में पेजरैंक कार्यान्वयन का एक स्थिर और गतिशील संस्करण है। स्थिर संस्करण में पुनरावृत्तियों की एक निश्चित संख्या होती है, जबकि गतिशील संस्करण तब तक काम करेगा जब तक कि रेटिंग दिए गए मूल्य में परिवर्तित होना शुरू न हो जाए।

हमारे ग्राफ के लिए, यह इस प्रकार होगा:

 def dynamicRanks(socialGraph: SocialGraph, tolerance: Double) = socialGraph.graph.pageRank(tol = tolerance).vertices def staticRanks(socialGraph: SocialGraph, tolerance: Double) = socialGraph.graph.staticPageRank(numIter = 20).vertices 

निष्कर्ष

एक चौकस पाठक ने नोट किया कि इस लेख का विषय रेखांकन के प्रसंस्करण को वितरित किया गया है, लेकिन कोड लिखते समय, हमने प्रसंस्करण को वास्तव में वितरित करने के लिए कुछ भी नहीं किया। और यहाँ हमें शुरुआत में ही एग्जेर डीजकस्ट्रा के उद्धरण को याद करना चाहिए। स्पार्क नाटकीय रूप से वितरित कंप्यूटिंग के बोझ और बोझ को लेकर हमारे जीवन को सरल बनाता है। एक वितरित क्लस्टर पर निष्पादित कोड लिखना इतना मुश्किल काम नहीं है जितना कि शुरुआत में लग सकता है। और यहां क्लस्टर संसाधनों के प्रबंधन के लिए और भी कई विकल्प हैं : Hadoop YARN, Apache Mesos (व्यक्तिगत रूप से, मेरा पसंदीदा विकल्प), और हाल ही में, कुबेरनेट्स के लिए समर्थन है। इस लेख में पार्स किए गए सभी स्रोत कोड को जीथब पर पाया जा सकता है।