स्पार्क स्ट्रीमिंग और काफ्का का एकीकरण

नमस्कार साथियों! हम आपको याद दिलाते हैं कि बहुत समय पहले हमने स्पार्क के बारे में एक पुस्तक प्रकाशित नहीं की थी, और अभी काफ्का के बारे में एक पुस्तक नवीनतम प्रूफरीडिंग से गुजर रही है।


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

1. औचित्य

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



2. काफ्का और स्पार्क स्ट्रीमिंग का एकीकरण

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



आइए इन तरीकों पर अधिक विस्तार से विचार करें।

एक। प्राप्तकर्ता आधारित दृष्टिकोण

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

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

इसके बाद, हम काफ्का - स्पार्क स्ट्रीमिंग के साथ एक आवेदन में प्राप्तकर्ताओं के साथ इस दृष्टिकोण का उपयोग करने का तरीका देखेंगे।

मैं। लिंक करना

अब हम अपने स्ट्रीमिंग एप्लिकेशन को स्काला / जावा अनुप्रयोगों के लिए निम्नलिखित विरूपण साक्ष्य से जोड़ेंगे, हम एसबीटी / मावेन के लिए प्रोजेक्ट परिभाषाओं का उपयोग करेंगे।

groupId = org.apache.spark artifactId = spark-streaming-kafka-0-8_2.11 version = 2.2.0 

हालांकि, हमारे आवेदन को तैनात करते समय, हमें उपरोक्त पुस्तकालय और उसकी निर्भरता को जोड़ना होगा, यह पायथन अनुप्रयोगों के लिए आवश्यक होगा।

ii। प्रोग्रामिंग

इसके बाद, स्ट्रीम एप्लिकेशन कोड में KafkaUtils को आयात करके एक DStream इनपुट स्ट्रीम बनाएं:

 import org.apache.spark.streaming.kafka._ val kafkaStream = KafkaUtils.createStream(streamingContext, [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume]) 

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

iii। तैनाती

किसी भी स्पार्क एप्लिकेशन के साथ, स्पार्क-सबमिट कमांड को लॉन्च करने के लिए उपयोग किया जाता है। हालाँकि, विवरण स्कैला / जावा अनुप्रयोगों और पायथन अनुप्रयोगों में थोड़ा अलग है।

इसके अलावा, –packages साथ आप spark-streaming-Kafka-0-8_2.11 जोड़ सकते हैं और इसकी निर्भरता सीधे spark-submit , यह पायथन एप्लिकेशन के लिए उपयोगी है जहां एसबीटी / मावेन का उपयोग करके परियोजनाओं का प्रबंधन करना असंभव है।

 ./bin/spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 ... 

आप मावेन रिपॉजिटरी से मावेन विरूपण साक्ष्य spark-streaming-Kafka-0-8-assembly का जार संग्रह भी डाउनलोड कर सकते हैं। फिर इसे spark-submit jars साथ spark-submit करें।

ख। प्रत्यक्ष दृष्टिकोण (कोई प्राप्तकर्ता नहीं)

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

नोट: यह सुविधा स्केला और जावा एपीआई के लिए स्पार्क 1.3 में दिखाई दी, साथ ही पायथन एपीआई के लिए स्पार्क 1.4 में भी।

अब चर्चा करते हैं कि हमारे स्ट्रीमिंग एप्लिकेशन में इस दृष्टिकोण को कैसे लागू किया जाए।
उपभोक्ता एपीआई निम्नलिखित लिंक पर अधिक विस्तार से वर्णित है:

अपाचे काफ्का उपभोक्ता | कफका उपभोक्ता के उदाहरण

मैं। लिंक करना

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

 groupId = org.apache.spark artifactId = spark-streaming-kafka-0-8_2.11 version = 2.2.0 

ii। प्रोग्रामिंग

अगला, KafkaUtils आयात करें और स्ट्रीम एप्लिकेशन कोड में एक इनपुट DStream बनाएं:

 import org.apache.spark.streaming.kafka._ val directKafkaStream = KafkaUtils.createDirectStream[ [key class], [value class], [key decoder class], [value decoder class] ]( streamingContext, [map of Kafka parameters], [set of topics to consume]) 

काफ्का मापदंडों में, आपको metadata.broker.list या bootstrap.servers निर्दिष्ट करने की आवश्यकता होगी। इसलिए, डिफ़ॉल्ट रूप से, हम काफ्का के प्रत्येक खंड में अंतिम ऑफसेट से शुरू होने वाले डेटा का उपभोग करेंगे। हालांकि, यदि आप चाहते हैं कि रीडिंग सबसे छोटे टुकड़े से शुरू हो, तो auto.offset.reset मापदंडों में आपको कॉन्फ़िगरेशन विकल्प auto.offset.reset सेट करना auto.offset.reset ।

इसके अलावा, KafkaUtils.createDirectStream विकल्पों के साथ काम करना, आप एक मनमाना ऑफसेट से पढ़ना शुरू कर सकते हैं। फिर हम निम्नलिखित कार्य करेंगे, जो हमें प्रत्येक पैकेज में खपत किए गए काफ्का अंशों तक पहुंचने की अनुमति देगा।

 //      ,        var offsetRanges = Array.empty[OffsetRange] directKafkaStream.transform { rdd => offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd }.map { ... }.foreachRDD { rdd => for (o <- offsetRanges) { println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}") } ... } 

यदि हम विशेष उपकरणों का उपयोग करके ज़ूकीपर के आधार पर काफ्का की निगरानी का आयोजन करना चाहते हैं, तो हम ज़ुकीपर को उनकी मदद से स्वयं अपडेट कर सकते हैं।

iii। तैनाती

इस मामले में तैनाती की प्रक्रिया प्राप्तकर्ता के साथ संस्करण में तैनाती प्रक्रिया जैसा दिखता है।

3. प्रत्यक्ष दृष्टिकोण का लाभ

कफ़्का के साथ स्पार्क स्ट्रीमिंग को एकीकृत करने के लिए दूसरा तरीका निम्नलिखित कारणों से पहला है।

एक। सरलीकृत संगरोध

इस मामले में, आपको कई काफ्का इनपुट स्ट्रीम बनाने और उन्हें संयोजित करने की आवश्यकता नहीं है। हालांकि, काफ्का - स्पार्क स्ट्रीमिंग कई आरडीडी सेगमेंट बनाएगी क्योंकि खपत के लिए काफ्का सेगमेंट होंगे। इस काफ़्का डेटा के सभी समानांतर में पढ़ा जाएगा। इसलिए, हम कह सकते हैं कि हमारे पास काफ्का और आरडीडी खंडों के बीच एक-से-एक पत्राचार होगा, और ऐसा मॉडल अधिक समझ में आता है और कॉन्फ़िगर करना आसान है।

ख। प्रभावशीलता

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

एक। बिल्कुल सही-एक बार शब्दार्थ

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

इसलिए, विफलताओं की स्थिति में भी, स्पार्क स्ट्रीमिंग प्रत्येक रिकॉर्ड को एक बार सख्ती से प्राप्त करता है। यहां हमें यह सुनिश्चित करने की आवश्यकता है कि हमारा आउटपुट ऑपरेशन, जिसमें डेटा बाहरी भंडारण में संग्रहीत है, या तो बेकार है या परमाणु लेनदेन है, जिसमें परिणाम और ऑफसेट दोनों संग्रहीत हैं। यह हमारे परिणामों की व्युत्पत्ति में ठीक-ठीक शब्दार्थ प्राप्त होता है।

हालाँकि, एक कमी है: ज़ुकाइपर में ऑफ़सेट अपडेट नहीं किए जाते हैं। इसलिए, ज़ूकीपर पर आधारित काफ्का के निगरानी उपकरण आपको प्रगति को ट्रैक करने की अनुमति नहीं देते हैं।
हालाँकि, हम अभी भी ऑफ़सेट का उल्लेख कर सकते हैं, अगर इस तरह से प्रोसेसिंग की व्यवस्था की जाती है - हम प्रत्येक पैकेज की ओर रुख करते हैं और ज़ुकाइपर को स्वयं अपडेट करते हैं।

हम सभी अपाचे काफ्का और स्पार्क स्ट्रीमिंग को एकीकृत करने के बारे में बात करना चाहते थे। हमें उम्मीद है कि आपको यह पसंद आया होगा।