किताब काफ्का स्ट्रीम्स इन एक्शन। वास्तविक समय अनुप्रयोगों और microservices »

हाय, हब्रोज़ितेली! यह पुस्तक किसी भी डेवलपर के लिए उपयुक्त है जो स्ट्रीमिंग प्रोसेसिंग को समझना चाहता है। वितरित प्रोग्रामिंग को समझने से आपको काफ्का और काफ्का धाराओं को बेहतर ढंग से समझने में मदद मिलेगी। काफ्का की रूपरेखा को जानना अच्छा होगा, लेकिन यह आवश्यक नहीं है: मैं आपको वह सब कुछ बताऊंगा जिसकी आपको आवश्यकता है। इस पुस्तक के लिए धन्यवाद, अनुभवी काफ्का डेवलपर्स, नौसिखियों की तरह, काफ्का धाराओं पुस्तकालय का उपयोग करके दिलचस्प स्ट्रीमिंग एप्लिकेशन बनाना सीखेंगे। इंटरमीडिएट और उच्च स्तर के जावा डेवलपर्स पहले से ही अवधारणाओं के साथ परिचित हैं जैसे कि धारावाहिकीकरण सीखेंगे कि काफ्का स्ट्रीम एप्लिकेशन बनाने के लिए अपने कौशल को कैसे लागू किया जाए। पुस्तक का स्रोत कोड जावा 8 में लिखा गया है और अनिवार्य रूप से जावा 8 के लैम्ब्डा अभिव्यक्तियों के वाक्य विन्यास का उपयोग करता है, इसलिए लैम्बडा फ़ंक्शन (यहां तक ​​कि एक अन्य प्रोग्रामिंग भाषा में) के साथ काम करने की क्षमता आपके लिए उपयोगी है।

अंश। 5.3। एकत्रीकरण और खिड़की संचालन

इस खंड में, हम काफ्का धाराओं के सबसे आशाजनक भागों की ओर बढ़ते हैं। अब तक हमने काफ्का धाराओं के निम्नलिखित पहलुओं को कवर किया है:

• एक प्रसंस्करण टोपोलॉजी बनाना;
• स्ट्रीमिंग अनुप्रयोगों में राज्य का उपयोग;
• डेटा स्ट्रीम कनेक्शन बनाना;
• घटना धाराओं (KStream) और अद्यतन धाराओं (KTable) के बीच अंतर।

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

5.3.1। उद्योग द्वारा स्टॉक की बिक्री का एकत्रीकरण

स्ट्रीमिंग डेटा के साथ काम करने के लिए एकत्रीकरण और समूहीकरण महत्वपूर्ण उपकरण हैं। प्रवेश के आधार पर व्यक्तिगत रिकॉर्ड की जांच करना अक्सर पर्याप्त नहीं होता है। डेटा से अतिरिक्त जानकारी निकालने के लिए, उनका समूहन और संयोजन आवश्यक है।

इस उदाहरण में, आपको एक इंट्राडे व्यापारी के सूट पर प्रयास करना होगा, जिसे कई उद्योगों में कंपनियों के शेयरों की बिक्री की मात्रा को ट्रैक करने की आवश्यकता है। विशेष रूप से, आप प्रत्येक उद्योग में सबसे बड़ी शेयर बिक्री के साथ पांच कंपनियों में रुचि रखते हैं।

इस तरह के एकत्रीकरण के लिए, आपको डेटा को वांछित रूप में (सामान्य शब्दों में) अनुवाद करने के लिए कई चरणों की आवश्यकता होगी।

1. एक विषय-आधारित स्रोत बनाएं जो कच्चे स्टॉक ट्रेडिंग की जानकारी प्रकाशित करता है। हमें ShareVolume के प्रकार की वस्तु के लिए StockTransaction का एक ऑब्जेक्ट मैप करना होगा। तथ्य यह है कि StockTransaction ऑब्जेक्ट में बिक्री मेटाडेटा है, और हमें केवल बेचे गए शेयरों की संख्या पर डेटा की आवश्यकता है।
2. स्टॉक प्रतीकों द्वारा ग्रुप शेयरवोल्यूम डेटा। प्रतीकों द्वारा समूहीकरण करने के बाद, आप इस डेटा को स्टॉक की बिक्री के सबटोटल्स में बदल सकते हैं। यह ध्यान देने योग्य है कि KStream.groupBy पद्धति KGroupedStream का एक प्रकार देता है। और आप बाद में KGroupedStream.reduce विधि को कॉल करके एक KTable उदाहरण प्राप्त कर सकते हैं।

KGroupedStream इंटरफ़ेस क्या है

KStream.groupBy और KStream.groupByKey विधियाँ KGroupedStream की एक आवृत्ति लौटाती हैं। KGroupedStream कुंजी द्वारा समूहीकृत करने के बाद ईवेंट स्ट्रीम का एक मध्यवर्ती प्रतिनिधित्व है। इसके साथ सीधे काम करने का इरादा बिल्कुल नहीं है। इसके बजाय, KGroupedStream का उपयोग एकत्रीकरण के संचालन के लिए किया जाता है, जिसका परिणाम हमेशा KTable होता है। और चूंकि एकत्रीकरण संचालन का परिणाम KTable है और वे राज्य भंडारण का उपयोग करते हैं, इसलिए संभव है कि पाइपलाइन के परिणामस्वरूप सभी अपडेट आगे नहीं भेजे जाएं।

KTable.groupBy विधि एक समान KGroupedTable लौटाती है - कुंजी द्वारा फिर से अपडेट किए गए अपडेट की धारा का एक मध्यवर्ती प्रतिनिधित्व।

आइए एक छोटा ब्रेक लें और अंजीर को देखें। 5.9, जो दिखाता है कि हमने क्या हासिल किया है। यह टोपोलॉजी आपको पहले से ही परिचित होना चाहिए।


अब इस टोपोलॉजी के लिए कोड पर एक नज़र डालते हैं (यह src / main / java / bbejeck / Chapter_5 / AggregationsAndReducingExample.java फ़ाइल में पाया जा सकता है) (लिस्टिंग 5.2)।


दिए गए कोड संक्षिप्तता और कई लाइनों में किए गए कार्यों की एक बड़ी मात्रा में भिन्न होते हैं। बिल्डर.स्ट्रीम विधि के पहले पैरामीटर में, आप अपने लिए कुछ नया नोटिस कर सकते हैं: प्रगणित प्रकार का मान AutoOffsetReset.EARLIEST (इसमें भी नवीनतम है), जिसका उपयोग करके .withOffsetResetPolicy विधि का उपयोग करके सेट किया गया है। इस गणना प्रकार का उपयोग करके, आप KStream या KTable में से प्रत्येक के लिए ऑफ़सेट रीसेट करने की रणनीति निर्दिष्ट कर सकते हैं; इसमें कॉन्फ़िगरेशन से ऑफ़सेट रीसेट करने के लिए पैरामीटर पर प्राथमिकता है।

GroupByKey और GroupBy

KStream इंटरफ़ेस में रिकॉर्डिंग के लिए दो तरीके हैं: GroupByKey और GroupBy। दोनों KGroupedTable को लौटाते हैं, इसलिए आपके पास एक वैध प्रश्न हो सकता है: उनके बीच क्या अंतर है और कब किसका उपयोग करना है?

GroupByKey विधि का उपयोग तब किया जाता है जब KStream में कुंजियाँ पहले से ही गैर-रिक्त हैं। और सबसे महत्वपूर्ण बात यह है कि झंडा "पुन: विभाजन की आवश्यकता है" कभी सेट नहीं किया गया।

GroupBy विधि मानती है कि आपने समूहन के लिए कुंजियाँ बदल दी हैं, इसलिए पुन: विभाजन ध्वज को सही पर सेट किया गया है। GroupBy विधि के बाद कनेक्शन, एकत्रीकरण, आदि प्रदर्शन करने से स्वचालित पुन: विभाजन हो जाएगा।
सारांश: आपको जब भी संभव हो GroupBy के बजाय GroupByKey का उपयोग करना चाहिए।

MapValues ​​और GroupBy विधियाँ क्या समझ में आती हैं, इसलिए योग () विधि पर नज़र डालें (यह src / main / java / bbejeck / model / ShareVolume.java फ़ाइल में मिल सकती है) (सूचीकरण 5.3)।


ShareVolume.sum विधि स्टॉक की बिक्री मात्रा का उप-योग लौटाती है, और संपूर्ण गणना श्रृंखला का परिणाम एक KTable <string, ShareVolume> ऑब्जेक्ट होता है। अब आप समझते हैं कि KTable की क्या भूमिका है। जब ShareVolume ऑब्जेक्ट आते हैं, तो नवीनतम वर्तमान अद्यतन इसी KTable में सहेजा जाता है। यह महत्वपूर्ण है कि यह न भूलें कि सभी अपडेट पिछले shareVolumeKTable में परिलक्षित होते हैं, लेकिन सभी को आगे नहीं भेजा जाता है।

इसके अलावा, इस KTable की सहायता से, हम प्रत्येक उद्योग में उच्चतम शेयर बिक्री के साथ पांच कंपनियों को प्राप्त करने के लिए एकत्रीकरण (बेचे गए शेयरों की संख्या) करते हैं। इस मामले में हमारी कार्रवाई पहले एकत्रीकरण के दौरान की जाने वाली क्रियाओं के समान होगी।

1. उद्योग द्वारा अलग-अलग ShareVolume वस्तुओं के समूह के लिए एक और GroupBy ऑपरेशन करें
2. ShareVolume ऑब्जेक्ट को सारांशित करने के लिए आगे बढ़ें। इस बार, एकत्रीकरण वस्तु एक निश्चित आकार की प्राथमिकता कतार है। बिकने वाली सबसे बड़ी संख्या वाली केवल पांच कंपनियों को इस तरह के एक निश्चित आकार की कतार में रखा जाता है।
3. एक स्ट्रिंग मान में पिछले पैराग्राफ से लाइनों को प्रदर्शित करें और उद्योग द्वारा शेयरों की संख्या से पांच सबसे अधिक बिक्री लौटाएं।
4. विषय के लिए स्ट्रिंग रूप में परिणाम लिखें।

अंजीर में। 5.10 डेटा आंदोलन की टोपोलॉजी का एक ग्राफ दिखाता है। जैसा कि आप देख सकते हैं, प्रसंस्करण का दूसरा दौर काफी सरल है।


अब, प्रसंस्करण के इस दूसरे दौर की संरचना को स्पष्ट रूप से समझने के बाद, आप इसके स्रोत कोड का उल्लेख कर सकते हैं (आप इसे फ़ाइल src / main / java / bbejeck / Chapter_5 / AggregationsAndReducingExample.java) (सूची 5.4) में देखेंगे।

इस इनिशियलाइज़र में फिक्स्ड क्यू वेरिएबल होता है। यह एक कस्टम ऑब्जेक्ट है - java.util.TreeSet के लिए एक एडेप्टर, जिसका उपयोग बेचे गए शेयरों की संख्या के घटते क्रम में एन उच्चतम परिणामों को ट्रैक करने के लिए किया जाता है।


आपको पहले से ही GroupBy और mapValues ​​पर कॉल का सामना करना पड़ा है, इसलिए हमने उन पर रोक नहीं लगाई है (हम KTable.toStream विधि को कॉल करते हैं, क्योंकि KTable.print विधि पदावनत है)। लेकिन आपने अभी तक एग्रीगेट () पद्धति के केटेबल संस्करण को नहीं देखा है, इसलिए हम इस पर चर्चा करने में कुछ समय व्यतीत करेंगे।

जैसा कि आपको याद है, KTable इस तथ्य से प्रतिष्ठित है कि एक ही कुंजी के साथ रिकॉर्ड को अपडेट माना जाता है। KTable नए के साथ पुराने रिकॉर्ड को बदल देता है। एकत्रीकरण उसी तरह से होता है: एक कुंजी के साथ पिछले रिकॉर्ड एकत्र होते हैं। जब कोई रिकॉर्ड आता है, तो यह एक एडिटर (एग्रीगेट विधि में कॉल में दूसरा पैरामीटर) का उपयोग करके फिक्स्डसाइज़पायरिटीक्यूयू वर्ग के एक उदाहरण में जोड़ा जाता है, लेकिन यदि उसी कुंजी के साथ एक और रिकॉर्ड पहले से मौजूद है, तो पुराना रिकॉर्ड सबट्रैक्टर (कॉल इन एग्रीगेट विधि में तीसरा पैरामीटर) का उपयोग करके हटा दिया जाता है।

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

हमने दो महत्वपूर्ण काम करना सीखा:

• KTable में समूह मान उनके लिए एक प्रमुख आम है;
• इन समूहित मूल्यों पर दृढ़ीकरण और एकत्रीकरण जैसे उपयोगी कार्य करें।

इन ऑपरेशनों को करने की क्षमता कफ़्का स्ट्रीम्स एप्लिकेशन के माध्यम से जाने वाले डेटा के अर्थ को समझने और यह पता लगाने के लिए महत्वपूर्ण है कि वे क्या जानकारी ले जाते हैं।

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

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

अगला, हम समय की एक निश्चित अवधि के भीतर - जैसे एकत्रीकरण के संचालन को देखते हैं, तथाकथित विंडो संचालन।

5.3.2। विंडो संचालन

पिछले अनुभाग में, हमने "रोलिंग" कन्वेंशन और एकत्रीकरण पेश किया। आवेदन ने पांच सबसे अधिक बिकने वाले शेयरों के बाद के एकत्रीकरण के साथ स्टॉक की बिक्री का एक निरंतर दृढ़ संकल्प किया।

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

खरीदार द्वारा विनिमय लेनदेन की गणना

निम्नलिखित उदाहरण में, हम कई व्यापारियों के लिए विनिमय लेनदेन पर नज़र रखने में संलग्न होंगे - या तो बड़े संगठन या स्मार्ट एकल-हाथ वाले फाइनेंसर।

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

ऐसी ट्रैकिंग में इस तरह के कदम शामिल हैं:

• स्टॉक-लेनदेन विषय से पढ़ने के लिए एक स्ट्रीम बनाना;
• ग्राहक आईडी और स्टॉक के प्रतीक द्वारा आने वाले रिकॉर्ड का समूहन। GroupBy विधि के लिए एक कॉल KGroupedStream वर्ग का एक उदाहरण देता है;
• KGroupedStream.windowedBy एक अस्थायी विंडो से घिरा डेटा स्ट्रीम देता है, जो विंडो एकत्रीकरण की अनुमति देता है। विंडो के प्रकार के आधार पर, TimeWindowedKStream या SessionWindowedKStream वापस कर दिया गया है;
• एक एकत्रीकरण ऑपरेशन के लिए लेनदेन की गणना। विंडो डेटा स्ट्रीम निर्धारित करता है कि क्या इस गणना में किसी विशेष रिकॉर्ड को ध्यान में रखा गया है;
• एक विषय के लिए परिणाम लिखना या विकास के दौरान उन्हें कंसोल में आउटपुट करना।

इस एप्लिकेशन की टोपोलॉजी सरल है, लेकिन इसकी दृश्य तस्वीर चोट नहीं करती है। तस्वीर को देखिए। 5.11।

इसके अलावा हम खिड़की के संचालन की कार्यक्षमता और संबंधित कोड पर विचार करेंगे।

खिड़की के प्रकार

कफ़्का धाराओं में तीन प्रकार की खिड़कियाँ हैं:

• सत्र;
• टम्बलिंग (टम्बलिंग);
• स्लाइडिंग / "कूद" (स्लाइडिंग / होपिंग)।

किसको चुनना है यह व्यवसाय की आवश्यकताओं पर निर्भर करता है। "टंबलिंग" और "जंपिंग" विंडो समय में सीमित हैं, जबकि सत्र प्रतिबंध उपयोगकर्ता कार्यों से जुड़े हैं - सत्र की अवधि उपयोगकर्ता द्वारा पूरी तरह से निर्धारित की जाती है कि उपयोगकर्ता कितनी सक्रियता से व्यवहार करता है। मुख्य बात यह नहीं भूलना है कि सभी प्रकार की खिड़कियां रिकॉर्ड्स की तारीख / समय टिकटों पर आधारित हैं, और सिस्टम समय पर नहीं।

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

सत्र खिड़कियां

सत्र विंडो अन्य सभी विंडो प्रकारों से बहुत अलग है। वे उपयोगकर्ता की गतिविधि (या इकाई की गतिविधि जिसे आप ट्रैक करना चाहते हैं) द्वारा समय के साथ इतना सीमित नहीं हैं। निष्क्रियता की अवधि के द्वारा सत्र विंडो को सीमांकित किया जाता है।

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

एक्सचेंज लेनदेन को ट्रैक करने के लिए सत्र विंडोज का उपयोग करना

हम विनिमय लेनदेन के बारे में जानकारी कैप्चर करने के लिए सत्र विंडो का उपयोग करेंगे। सत्र विंडोज़ का कार्यान्वयन सूची 5.5 में दिखाया गया है (जो src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKTableJoinExample.java में पाया जा सकता है)।


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

किसी भी GroupBy ऑपरेशन के लिए, कुछ प्रकार के एकत्रीकरण ऑपरेशन (एकत्रीकरण, दृढ़ संकल्प या गिनती) आमतौर पर किए जाते हैं। आप संचयी कुल या विंडो एकत्रीकरण के साथ या तो संचयी एकत्रीकरण कर सकते हैं, जिसमें दिए गए समय विंडो के भीतर रिकॉर्ड को ध्यान में रखा जाता है।

5.5 लिस्टिंग में कोड सत्र खिड़कियों के भीतर लेनदेन की संख्या मायने रखता है। अंजीर में। 5.13 इन क्रियाओं का चरण दर चरण विश्लेषण किया जाता है।

WindowedBy (SessionWindows.with (बीस सेकंड) .until (पंद्रहMinutes) कॉल करके हम 20 सेकंड के निष्क्रिय अंतराल और 15 मिनट के प्रतिधारण अंतराल के साथ एक सत्र विंडो बनाते हैं। 20 सेकंड के निष्क्रियता अंतराल का मतलब है कि आवेदन में वर्तमान सत्र (सक्रिय) सत्र में वर्तमान सत्र के अंत या शुरुआत से 20 सेकंड के भीतर आने वाला कोई भी रिकॉर्ड शामिल होगा।


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

आइए गिनती पद्धति से कुछ प्रविष्टियों को देखें कि सत्र कैसे काम करता है (तालिका 5.1)।


रिकॉर्ड प्राप्त होने पर, हम पहले से ही एक ही कुंजी के साथ मौजूदा सत्रों की तलाश करते हैं, अंत समय वर्तमान दिनांक / समय स्टाम्प से कम है - निष्क्रियता अंतराल और प्रारंभ समय वर्तमान तिथि / समय स्टाम्प + निष्क्रियता अंतराल से अधिक है। इसे ध्यान में रखते हुए, तालिका से चार रिकॉर्ड। 5.1 निम्नानुसार एक सत्र में विलय।

1. रिकॉर्ड 1 पहले आता है, इसलिए प्रारंभ समय अंत समय के बराबर है और 00:00:00 है।

2. अगला रिकॉर्ड 2 आता है, और हम सत्रों की तलाश करते हैं जो 23:59:55 से पहले समाप्त नहीं होते हैं और बाद में 00:00:35 से शुरू नहीं होते हैं। रिकॉर्ड 1 खोजें और सत्र 1 और 2 को मिलाएं। सत्र 1 (पहले) और सत्र 2 के अंत समय (बाद में) का प्रारंभ समय लें, ताकि हमारा नया सत्र 00:00:00 बजे शुरू हो और 00:00:15 बजे समाप्त हो।

3. रिकॉर्ड 3 आता है, हम 00:00:30 और 00:01:10 के बीच सत्रों की तलाश करते हैं और कोई नहीं पाते हैं। कुंजी के लिए दूसरा सत्र जोड़ें 123-345-654, FFBE, शुरू और अंत 00:00:50 पर।

4. रिकॉर्ड 4 आता है, और हम 23:59:45 और 00:00:25 के बीच सत्रों की तलाश करते हैं। इस बार दोनों सत्र हैं - 1 और 2. सभी तीन सत्रों को एक में जोड़ा जाता है, जिसमें 00:00:00 का प्रारंभ समय और 00:00:15 का अंत समय होता है।

इस खंड में क्या कहा गया है, यह निम्नलिखित महत्वपूर्ण बारीकियों को याद रखने योग्य है:

• सत्र निश्चित आकार की खिड़कियां नहीं हैं। एक सत्र की अवधि दी गई अवधि के भीतर गतिविधि द्वारा निर्धारित की जाती है;
• डेटा में दिनांक / समय टिकटें यह निर्धारित करती हैं कि कोई घटना मौजूदा सत्र में आती है या निष्क्रियता की अवधि में।

आगे हम निम्न प्रकार की खिड़कियों पर चर्चा करेंगे - "somersault" विंडोज़।

टंबलिंग विंडो

"टंबलिंग" विंडोज़ एक निश्चित अवधि के भीतर होने वाली घटनाओं को पकड़ लेती है। कल्पना करें कि आपको हर 20 सेकंड में एक कंपनी के सभी विनिमय लेनदेन को पकड़ने की आवश्यकता है, ताकि आप इस अवधि के लिए सभी घटनाओं को इकट्ठा कर सकें। 20-सेकंड के अंतराल के अंत में, विंडो "tumbles" और एक नए 20-सेकंड के अवलोकन अंतराल पर स्विच करता है। चित्र 5.14 इस स्थिति को दर्शाता है।


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

5.6 को सूचीबद्ध करना कोड को दिखाता है जो हर 20 सेकंड में विनिमय लेनदेन पर कब्जा करने के लिए टंबलिंग विंडो का उपयोग करता है (आप इसे src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKtableJoinExample.java में पा सकते हैं)।


TimeWindows.of पद्धति पर कॉल करने के लिए इस छोटे से बदलाव के लिए धन्यवाद, आप टंबलिंग विंडो का उपयोग कर सकते हैं। इस उदाहरण में, तब तक () विधि के लिए कोई कॉल नहीं है, जिसके परिणामस्वरूप 24 घंटे के डिफ़ॉल्ट सहेजें अंतराल का उपयोग किया जाएगा।

अंत में, यह खिड़की के विकल्पों में से अंतिम पर जाने का समय है - खिड़कियों को बंद करना।

स्लाइडिंग ("कूद") खिड़कियां

स्लाइडिंग / "hopping" खिड़कियां "tumbling" के समान हैं, लेकिन थोड़े अंतर के साथ। स्लाइडिंग विंडो हाल की घटनाओं को संभालने के लिए एक नई विंडो बनाने से पहले समय अंतराल के अंत की प्रतीक्षा नहीं करती है। वे विंडो की अवधि से कम प्रतीक्षा अंतराल के बाद नई गणना शुरू करते हैं।

"सोमरसॉल्टिंग" और "जंपिंग" विंडो के बीच के अंतर को स्पष्ट करने के लिए, चलिए विनिमय लेनदेन की गणना के साथ उदाहरण पर लौटते हैं। हमारा लक्ष्य, पहले की तरह, लेन-देन की संख्या गिनना है, लेकिन हम काउंटर अपडेट करने से पहले हर समय इंतजार नहीं करना चाहेंगे। इसके बजाय, हम छोटे अंतराल पर काउंटर को अपडेट करेंगे। उदाहरण के लिए, हम हर 20 सेकंड में लेन-देन की संख्या की गणना करना जारी रखेंगे, लेकिन प्रत्येक 5 सेकंड में काउंटर को अपडेट करने के लिए, जैसा कि अंजीर में दिखाया गया है। 5.15। उसी समय, हमारे पास अतिव्यापी डेटा के साथ तीन परिणाम विंडो हैं।


5.7 को सूचीबद्ध करना फिसलने वाली खिड़कियों को निर्दिष्ट करने के लिए कोड दिखाता है (यह src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKtableJoinExample.java में पाया जा सकता है)।


एडवांसबाय () पद्धति पर कॉल जोड़कर एक टंबलिंग विंडो को टंबलिंग विंडो में बदला जा सकता है। उपर्युक्त उदाहरण में, सेव अंतराल 15 मिनट है।

आपने इस खंड में देखा कि समय खिड़कियों पर एकत्रीकरण के परिणामों को कैसे सीमित किया जाए। विशेष रूप से, मैं चाहूंगा कि आप इस खंड से निम्नलिखित तीन बातें याद रखें:

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

अगला, हम सीखेंगे कि कनेक्शन के लिए KTable को KStream में कैसे बदलें।

5.3.3। KStream और KTable ऑब्जेक्ट्स से जुड़ें

अध्याय 4 में, हमने दो KStream ऑब्जेक्ट्स को जोड़ने पर चर्चा की। अब हमें यह सीखना है कि KTable और KStream को कैसे जोड़ा जाए। निम्नलिखित सरल कारण के लिए इसकी आवश्यकता हो सकती है। KStream अभिलेखों की एक धारा है, और KTable रिकॉर्ड अद्यतनों की एक धारा है, लेकिन कभी-कभी KTable के अद्यतनों का उपयोग करके रिकॉर्ड की धारा में अतिरिक्त संदर्भ जोड़ना आवश्यक हो सकता है।

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

1. एक्सचेंजों की संख्या पर डेटा के साथ एक KTable ऑब्जेक्ट को KStream में कनवर्ट करें, इसके बाद कुंजी को उस कुंजी के साथ बदल दें जो इस शेयर के प्रतीक के अनुरूप उद्योग की पहचान करता है।
2. एक KTable ऑब्जेक्ट बनाएं जो किसी विषय के डेटा को विनिमय समाचार के साथ पढ़ता है। यह नया KTable उद्योग द्वारा वर्गीकृत किया जाएगा।
3. उद्योग द्वारा विनिमय लेनदेन की संख्या की जानकारी के साथ समाचार अपडेट को मिलाएं।

अब देखते हैं कि इस कार्य योजना को कैसे लागू किया जाए।

कन्वर्ट KTable को KStream

KTable को KStream में बदलने के लिए, निम्न कार्य करें।

1. KTable.toStream () विधि को कॉल करें।
2. KStream.map विधि कॉल का उपयोग करके, उद्योग के नाम के साथ कुंजी को बदलें, और फिर विंडोेड उदाहरण से TransactionSummary ऑब्जेक्ट निकालें।

हम इन कार्यों को निम्न प्रकार से श्रृंखलाबद्ध करते हैं (कोड src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKtableJoinExample.java में पाया जा सकता है) (लिस्टिंग 5.8)।


चूंकि हम KStream.map ऑपरेशन करते हैं, इसलिए कनेक्शन में उपयोग किए जाने पर लौटे KStream उदाहरण के लिए बार-बार विभाजन स्वचालित रूप से किया जाता है।

हमने रूपांतरण प्रक्रिया पूरी कर ली है, फिर हमें स्टॉक समाचार पढ़ने के लिए एक KTable ऑब्जेक्ट बनाने की आवश्यकता है।

शेयर बाजार समाचार के लिए KTable बनाना

सौभाग्य से, कोड की एक एकल लाइन एक KTable ऑब्जेक्ट बनाने के लिए पर्याप्त है (यह कोड src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKtableJoinExample.java) (लिस्टिंग 5.9) में पाया जा सकता है।


यह ध्यान देने योग्य है कि आपको किसी भी सर्ड ऑब्जेक्ट को निर्दिष्ट करने की आवश्यकता नहीं है, क्योंकि सेटिंग्स स्ट्रिंग सर्ड का उपयोग करती हैं। इसके अलावा, EARLIEST गणना का उपयोग करके, तालिका बहुत शुरुआत में प्रविष्टियों के साथ आबादी है।

अब हम अंतिम चरण - संघ पर आगे बढ़ सकते हैं।

लेन-देन संख्याओं के लिए समाचार अपडेट लिंक करें

संबंध बनाना सीधा है। हम संबंधित उद्योग के लिए कोई विनिमय समाचार नहीं होने की स्थिति में बाएं कनेक्शन का उपयोग करेंगे (आवश्यक कोड फाइल src / main / java / bbejeck / Chapter_5 / CountingWindowingAndKtableJoinExample .java) में पाया जा सकता है (लिस्टिंग 5.10)।


यह बायाँ कथन काफी सरल है। अध्याय 4 में कनेक्शन के विपरीत, JoinWindow विधि का उपयोग नहीं किया जाता है, क्योंकि जब आप KStream-KTable बनाते हैं, तो KTable में प्रत्येक कुंजी के लिए केवल एक रिकॉर्ड होता है। ऐसा कनेक्शन समय में सीमित नहीं है: रिकॉर्ड या तो केटेबल में है या अनुपस्थित है। मुख्य निष्कर्ष: KTable वस्तुओं की मदद से, आप कम अक्सर अद्यतन किए गए संदर्भ डेटा के साथ KStream को समृद्ध कर सकते हैं।

अब हम KStream से घटनाओं को समृद्ध करने के लिए अधिक कुशल तरीके से देखेंगे।

5.3.4। GlobalKTable वस्तुएँ

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

पुन: विभाजन एक मूल्य पर आता है

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

छोटे डेटासेट से कनेक्ट करें

कुछ मामलों में, कनेक्शन की योजना बनाई गई डेटा की मात्रा अपेक्षाकृत कम है, ताकि उनमें से पूरी प्रतियां स्थानीय रूप से प्रत्येक नोड पर फिट हो सकें। ऐसी स्थितियों के लिए, काफ्का स्ट्रीम्स GlobalKTable वर्ग प्रदान करता है।

GlobalKTable इंस्टेंसेस अद्वितीय हैं क्योंकि एप्लिकेशन प्रत्येक डेटा को प्रत्येक नोड में दोहराता है। और चूंकि सभी डेटा प्रत्येक नोड पर मौजूद हैं, इसलिए संदर्भ डेटा कुंजी द्वारा घटनाओं की धारा को विभाजित करने की कोई आवश्यकता नहीं है ताकि यह सभी वर्गों के लिए सुलभ हो। GlobalKTable ऑब्जेक्ट भी बिना कनेक्शन के प्रदर्शन कर सकते हैं। आइए इस सुविधा को प्रदर्शित करने के लिए पिछले उदाहरणों में से एक पर वापस जाएं।

KStream को GlobalKTable ऑब्जेक्ट से कनेक्ट करना

उपसमूह 5.3.2 में, हमने ग्राहकों द्वारा विनिमय लेनदेन के विंडो एकत्रीकरण का प्रदर्शन किया। इस एकत्रीकरण के परिणाम कुछ इस तरह दिखे:

{customerId='074-09-3705', stockTicker='GUTM'}, 17 {customerId='037-34-5184', stockTicker='CORK'}, 16 

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

ऐसा करने के लिए, हम काउंटस्ट्रीम ऑब्जेक्ट को Listing 5.11 (इसी कोड को src / main / java / bbejeck / Chapter_5 / GlobalKTableExample.java फ़ाइल में पाया जा सकता है) से इस्तेमाल करेंगे, इसे दो GlobalKTable ऑब्जेक्ट्स से जोड़ते हुए।


हम पहले भी इस पर चर्चा कर चुके हैं, इसलिए मैंने इसे दोहराया नहीं। लेकिन मैं ध्यान देता हूं कि कोड इनस्ट्रीम () में है। मैप किए गए फ़ंक्शन को एंबेडेड लैम्ब्डा एक्सप्रेशन के बजाय पठनीयता के लिए फ़ंक्शन ऑब्जेक्ट में एब्स्ट्रैक्ट किया गया है।

अगला कदम GlobalKTable के दो उदाहरणों की घोषणा करना है (दिखाया गया कोड src / main / java / bbejeck / Chapter_5 / GlobalKTableExample.java में पाया जा सकता है) (लिस्टिंग 5.12)।



ध्यान दें कि विषय के नाम प्रगणित प्रकारों का उपयोग करके वर्णित हैं।

अब जब हमने सभी घटक तैयार कर लिए हैं, तो यह कनेक्शन के लिए कोड लिखने के लिए बना हुआ है (जो कि src / main / java / bbejeck / Chapter_5 / GlobalKTableExample.java फ़ाइल में पाया जा सकता है) (लिस्टिंग 5.13)।


हालांकि इस कोड में दो यौगिक हैं, वे एक श्रृंखला में आयोजित किए जाते हैं, क्योंकि उनके किसी भी परिणाम का अलग से उपयोग नहीं किया जाता है। पूरे ऑपरेशन के अंत में परिणाम प्रदर्शित किए जाते हैं।

जब आप उपरोक्त कनेक्शन ऑपरेशन शुरू करते हैं, तो आपको निम्नलिखित परिणाम मिलेंगे:

 {customer='Barney, Smith' company="Exxon", transactions= 17} 

सार बदल नहीं गया है, लेकिन ये परिणाम अधिक स्पष्ट दिखते हैं।

अध्याय 4 की गिनती, आपने पहले ही कार्रवाई में कई प्रकार के कनेक्शन देखे हैं। वे तालिका में सूचीबद्ध हैं। 5.2। यह तालिका Kafka धाराओं के संस्करण 1.0.0 के लिए प्रासंगिक कनेक्टिविटी को दर्शाती है; भविष्य के रिलीज में कुछ बदल जाएगा।


अंत में, मैं आपको मुख्य बात याद दिलाऊंगा: आप स्थानीय राज्य का उपयोग करके इवेंट स्ट्रीम (KStream) और अपडेट स्ट्रीम (KTable) कनेक्ट कर सकते हैं। इसके अलावा, यदि संदर्भ डेटा का आकार बहुत बड़ा नहीं है, तो आप GlobalKTable ऑब्जेक्ट का उपयोग कर सकते हैं। GlobalKTable, Kafka Streams एप्लिकेशन के प्रत्येक नोड के सभी अनुभागों की प्रतिकृति करता है, जिससे सभी डेटा की उपलब्धता सुनिश्चित होती है, चाहे जिस सेक्शन से संबंधित हो।

आगे हम काफ्का धाराओं की संभावना देखेंगे, जिसकी बदौलत आप काफ्का विषय के आंकड़ों का उपभोग किए बिना राज्य में बदलाव देख सकते हैं।

5.3.5। अनुरोध की स्थिति

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

इन विषयों से डेटा पढ़ना एक तरह का भौतिक विचार माना जा सकता है। हमारे कार्यों के लिए, हम विकिपीडिया से भौतिकवादी दृष्टिकोण की परिभाषा का उपयोग कर सकते हैं: "... एक भौतिक डेटाबेस ऑब्जेक्ट जिसमें क्वेरी के परिणाम होते हैं। उदाहरण के लिए, यह हटाए गए डेटा की एक स्थानीय प्रतिलिपि हो सकती है, या किसी तालिका की पंक्तियों और / या स्तंभों का एक उपसमुच्चय हो सकता है या परिणामों में शामिल हो सकता है, या एकत्रीकरण का उपयोग करके प्राप्त एक धुरी तालिका "(https://en.wikipedia.org/wiki/Materialized-view) हो सकता है।

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

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

• डेटा की स्थानीयता के कारण, आप उन्हें जल्दी से एक्सेस कर सकते हैं
• डेटा के दोहराव को बाहर रखा गया है, क्योंकि वे बाहरी भंडारण के लिए नहीं लिखे गए हैं।

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

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

सारांश

• KStream ऑब्जेक्ट डेटाबेस आवेषण के लिए तुलनीय घटना धाराओं का प्रतिनिधित्व करते हैं। KTable ऑब्जेक्ट्स अपडेट स्ट्रीम का प्रतिनिधित्व करते हैं, वे डेटाबेस में अपडेट के समान होते हैं। KTable ऑब्जेक्ट का आकार नहीं बढ़ता है, पुराने रिकॉर्ड को नए लोगों के साथ बदल दिया जाता है।
• एकत्रीकरण कार्यों के लिए KTable वस्तुओं की आवश्यकता होती है।
• विंडो ऑपरेशंस का उपयोग करके, आप एकत्रित डेटा को समय की टोकरियों में तोड़ सकते हैं।
• GlobalKTable ऑब्जेक्ट्स के लिए धन्यवाद, आप सेक्शनिंग की परवाह किए बिना एप्लिकेशन में कहीं भी संदर्भ डेटा एक्सेस कर सकते हैं।
• KStream, KTable और GlobalKTable वस्तुओं के बीच संबंध संभव हैं।

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

→ पुस्तक का अधिक विवरण प्रकाशक की वेबसाइट पर पाया जा सकता है

→ खाबरोज़ीते के लिए कूपन पर 25% की छूट - काफ्का स्ट्रीम

→ पुस्तक के पेपर संस्करण के भुगतान पर, एक इलेक्ट्रॉनिक पुस्तक ई-मेल द्वारा भेजी जाती है।