"काफ्का स्ट्रीम्स इन अॅक्शन" हे पुस्तक. रिअल-टाइम कामासाठी ऍप्लिकेशन्स आणि मायक्रो सर्व्हिसेस"

हॅलो, खबरो रहिवासी! थ्रेड प्रोसेसिंग समजून घेऊ इच्छिणाऱ्या कोणत्याही विकसकासाठी हे पुस्तक योग्य आहे. वितरित प्रोग्रामिंग समजून घेतल्याने तुम्हाला काफ्का आणि काफ्का प्रवाह अधिक चांगल्या प्रकारे समजून घेण्यात मदत होईल. काफ्का फ्रेमवर्क स्वतः जाणून घेणे छान होईल, परंतु हे आवश्यक नाही: मी तुम्हाला आवश्यक ते सर्व सांगेन. काफ्का स्ट्रीम्स लायब्ररीचा वापर करून मनोरंजक स्ट्रीम प्रोसेसिंग अॅप्लिकेशन्स कसे तयार करायचे हे अनुभवी काफ्का डेव्हलपर आणि नवशिक्या या पुस्तकात शिकतील. सीरियलायझेशनसारख्या संकल्पनांशी आधीच परिचित असलेले इंटरमीडिएट आणि प्रगत Java डेव्हलपर काफ्का स्ट्रीम्स अॅप्लिकेशन्स तयार करण्यासाठी त्यांची कौशल्ये लागू करण्यास शिकतील. पुस्तकाचा सोर्स कोड Java 8 मध्ये लिहिलेला आहे आणि Java 8 lambda एक्स्प्रेशन सिंटॅक्सचा लक्षणीय वापर करतो, त्यामुळे lambda फंक्शन्स (अगदी दुसर्‍या प्रोग्रामिंग भाषेतही) कसे कार्य करायचे हे जाणून घेणे उपयुक्त ठरेल.

उतारा. ५.३. एकत्रीकरण आणि विंडो ऑपरेशन

या विभागात, आम्ही काफ्का प्रवाहांचे सर्वात आशादायक भाग एक्सप्लोर करण्यासाठी पुढे जाऊ. आतापर्यंत आम्ही काफ्का प्रवाहांच्या खालील पैलूंचा समावेश केला आहे:

• प्रक्रिया टोपोलॉजी तयार करणे;
• स्ट्रीमिंग ऍप्लिकेशन्समध्ये राज्य वापरणे;
• डेटा प्रवाह कनेक्शन करत आहे;
• इव्हेंट प्रवाह (KStream) आणि अद्यतन प्रवाह (KTable) मधील फरक.

पुढील उदाहरणांमध्ये आपण हे सर्व घटक एकत्र आणू. स्ट्रीमिंग ऍप्लिकेशन्सचे आणखी एक उत्कृष्ट वैशिष्ट्य, आपण विंडोिंगबद्दल देखील जाणून घ्याल. आमचे पहिले उदाहरण एक साधे एकत्रीकरण असेल.

५.३.१. उद्योग क्षेत्राद्वारे स्टॉक विक्रीचे एकत्रीकरण

प्रवाहित डेटासह कार्य करताना एकत्रीकरण आणि गट करणे ही महत्त्वपूर्ण साधने आहेत. वैयक्तिक नोंदी प्राप्त झाल्यामुळे त्यांची तपासणी अनेकदा अपुरी असते. डेटामधून अतिरिक्त माहिती काढण्यासाठी, त्यांना गटबद्ध करणे आणि एकत्र करणे आवश्यक आहे.

या उदाहरणात, तुम्ही एका दिवसाच्या व्यापार्‍याचा पोशाख घालाल ज्याला अनेक उद्योगांमधील कंपन्यांच्या समभागांच्या विक्रीच्या प्रमाणाचा मागोवा घेणे आवश्यक आहे. विशेषत:, प्रत्येक उद्योगात सर्वाधिक शेअर विक्री असलेल्या पाच कंपन्यांमध्ये तुम्हाला स्वारस्य आहे.

डेटाचे इच्छित फॉर्ममध्ये (सामान्य शब्दात बोलणे) भाषांतर करण्यासाठी अशा एकत्रीकरणासाठी खालील अनेक चरणांची आवश्यकता असेल.

1. कच्चा स्टॉक ट्रेडिंग माहिती प्रकाशित करणारा विषय-आधारित स्रोत तयार करा. शेअर व्हॉल्यूम प्रकारातील ऑब्जेक्टवर स्टॉक ट्रॅन्सॅक्शन प्रकारातील ऑब्जेक्ट मॅप करावा लागेल. मुद्दा असा आहे की स्टॉक ट्रान्झॅक्शन ऑब्जेक्टमध्ये विक्री मेटाडेटा आहे, परंतु आम्हाला फक्त विकल्या जाणार्‍या समभागांच्या संख्येबद्दल डेटा आवश्यक आहे.
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).

दिलेला कोड त्याच्या संक्षिप्ततेने आणि अनेक ओळींमध्ये केलेल्या क्रियांच्या मोठ्या प्रमाणात ओळखला जातो. तुम्हाला builder.stream पद्धतीच्या पहिल्या पॅरामीटरमध्ये काहीतरी नवीन दिसू शकते: Enum प्रकार AutoOffsetReset.EARLIEST (तेथे एक नवीनतम देखील आहे), Consumed.withOffsetResetPolicy पद्धत वापरून सेट केलेले मूल्य. या गणनेचा प्रकार प्रत्येक KStream किंवा KTable साठी ऑफसेट रीसेट धोरण निर्दिष्ट करण्यासाठी वापरला जाऊ शकतो आणि कॉन्फिगरेशनमधील ऑफसेट रीसेट पर्यायापेक्षा प्राधान्य देतो.

GroupByKey आणि GroupBy

KStream इंटरफेसमध्ये रेकॉर्ड गटबद्ध करण्यासाठी दोन पद्धती आहेत: GroupByKey आणि GroupBy. दोघेही KGroupedTable परत करतात, त्यामुळे तुम्ही विचार करत असाल की त्यांच्यात काय फरक आहे आणि कोणता कधी वापरायचा?

GroupByKey पद्धत वापरली जाते जेव्हा KStream मधील की आधीच रिकामी नसतात. आणि सर्वात महत्त्वाचे म्हणजे, "पुन्हा विभाजन करणे आवश्यक आहे" ध्वज कधीही सेट केला गेला नाही.

GroupBy पद्धत असे गृहीत धरते की तुम्ही ग्रुपिंग की बदलल्या आहेत, त्यामुळे पुनर्विभाजन ध्वज सत्यावर सेट केला आहे. ग्रुपबाय पद्धतीनंतर जोडणे, एकत्रीकरण इ. केल्याने स्वयंचलित री-विभाजन होईल.
सारांश: जेव्हा शक्य असेल तेव्हा तुम्ही GroupBy ऐवजी GroupByKey वापरावे.

mapValues ​​आणि groupBy पद्धती काय करतात हे स्पष्ट आहे, तर चला sum() पद्धत (src/main/java/bbejeck/model/ShareVolume.java मध्ये आढळते) (लिस्टिंग 5.3) वर एक नजर टाकूया.

ShareVolume.sum पद्धत स्टॉक विक्री व्हॉल्यूमची चालू असलेली एकूण रक्कम परत करते आणि संपूर्ण गणना साखळीचा परिणाम KTable ऑब्जेक्ट आहे . आता तुम्हाला KTable ची भूमिका समजली आहे. जेव्हा ShareVolume ऑब्जेक्ट्स येतात, तेव्हा संबंधित KTable ऑब्जेक्ट नवीनतम वर्तमान अपडेट संग्रहित करते. हे लक्षात ठेवणे महत्त्वाचे आहे की सर्व अद्यतने मागील shareVolumeKTable मध्ये प्रतिबिंबित होतात, परंतु सर्व पुढे पाठवले जात नाहीत.

पुढे, या KTable चा वापर करून, आम्ही एकत्रितपणे (व्यापार केलेल्या शेअर्सच्या संख्येनुसार) प्रत्येक उद्योगात सर्वाधिक प्रमाणात शेअर्सचा व्यापार करणाऱ्या पाच कंपन्यांकडे पोहोचतो. या प्रकरणातील आमच्या कृती पहिल्या एकत्रीकरणासारख्याच असतील.

1. इंडस्ट्रीनुसार वैयक्तिक शेअर व्हॉल्यूम ऑब्जेक्ट्सचे गट करण्यासाठी दुसरे ग्रुपबाय ऑपरेशन करा.
2. शेअर व्हॉल्यूम ऑब्जेक्ट्सचा सारांश देणे सुरू करा. यावेळी एकत्रीकरण ऑब्जेक्ट एक निश्चित-आकाराची प्राधान्य रांग आहे. या निश्चित-आकाराच्या रांगेत, सर्वात जास्त शेअर्स विकल्या गेलेल्या फक्त पाच कंपन्याच ठेवल्या जातात.
3. मागील परिच्छेदापासून स्ट्रिंग व्हॅल्यूवर रांगांचा नकाशा बनवा आणि उद्योगानुसार क्रमांकानुसार सर्वात जास्त व्यापार केलेले शीर्ष पाच स्टॉक परत करा.
4. विषयावर परिणाम स्ट्रिंग स्वरूपात लिहा.

अंजीर मध्ये. आकृती 5.10 डेटा फ्लो टोपोलॉजी आलेख दाखवते. जसे आपण पाहू शकता, प्रक्रियेची दुसरी फेरी अगदी सोपी आहे.

आता आम्हाला प्रक्रियेच्या या दुस-या फेरीची रचना स्पष्टपणे समजली आहे, आम्ही त्याच्या स्त्रोत कोडकडे वळू शकतो (तुम्हाला ते src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java या फाईलमध्ये सापडेल) (लिस्टिंग 5.4) .

या इनिशिएलायझरमध्ये एक निश्चित रांग व्हेरिएबल आहे. हा एक सानुकूल ऑब्जेक्ट आहे जो java.util.TreeSet साठी अॅडॉप्टर आहे जो व्यापार केलेल्या शेअर्सच्या उतरत्या क्रमाने शीर्ष N परिणामांचा मागोवा घेण्यासाठी वापरला जातो.

तुम्ही आधीच groupBy आणि mapValues ​​कॉल पाहिले आहेत, त्यामुळे आम्ही त्यामध्ये जाणार नाही (आम्ही KTable.toStream पद्धतीला कॉल करत आहोत कारण KTable.print पद्धत बहिष्कृत आहे). परंतु तुम्ही अद्याप aggregate() ची KTable आवृत्ती पाहिली नाही, म्हणून आम्ही त्यावर चर्चा करण्यासाठी थोडा वेळ घालवू.

तुम्हाला आठवत असेल, KTable ला काय वेगळे बनवते ते म्हणजे समान की सह रेकॉर्ड अद्यतने मानले जातात. केटेबल जुनी नोंद बदलून नवीन नोंदवते. एकत्रीकरण अशाच प्रकारे होते: समान की सह नवीनतम रेकॉर्ड एकत्रित केले जातात. जेव्हा एखादा रेकॉर्ड येतो, तेव्हा तो अॅडर वापरून FixedSizePriorityQueue वर्ग उदाहरणामध्ये जोडला जातो (एकूण पद्धत कॉलमधील दुसरा पॅरामीटर), परंतु त्याच कीसह दुसरा रेकॉर्ड आधीपासूनच अस्तित्वात असल्यास, वजाबाकी वापरून जुने रेकॉर्ड काढून टाकले जाते (तिसरे पॅरामीटर एकूण पद्धत कॉल).

या सर्वांचा अर्थ असा आहे की आमचा एग्रीगेटर, FixedSizePriorityQueue, सर्व मूल्ये एका कीसह एकत्रित करत नाही, परंतु N सर्वात जास्त ट्रेड केलेल्या स्टॉक्सच्या प्रमाणांची एकत्रित रक्कम संग्रहित करतो. प्रत्येक इनकमिंग एंट्रीमध्ये आतापर्यंत विकलेल्या शेअर्सची एकूण संख्या असते. KTable तुम्हाला प्रत्येक अपडेटच्या रोलिंग एग्रीगेशनची आवश्यकता न ठेवता सध्या कोणत्या कंपन्यांच्या शेअर्सची सर्वाधिक खरेदी-विक्री आहे याची माहिती देईल.

आम्ही दोन महत्त्वाच्या गोष्टी करायला शिकलो:

• सामान्य की द्वारे KTable मधील मूल्ये गट करा;
• या गटबद्ध मूल्यांवर रोलअप आणि एकत्रीकरण यासारखी उपयुक्त क्रिया करा.

काफ्का स्ट्रीम्स ऍप्लिकेशनमधून फिरणाऱ्या डेटाचा अर्थ समजून घेण्यासाठी आणि त्यात कोणती माहिती आहे हे समजून घेण्यासाठी ही ऑपरेशन्स कशी करावी हे जाणून घेणे महत्त्वाचे आहे.

आम्ही या पुस्तकात आधी चर्चा केलेल्या काही प्रमुख संकल्पना देखील एकत्र आणल्या आहेत. प्रकरण 4 मध्ये, आम्ही स्ट्रीमिंग ऍप्लिकेशनसाठी दोष-सहिष्णु, स्थानिक राज्य किती महत्त्वाचे आहे यावर चर्चा केली. या प्रकरणातील पहिल्या उदाहरणाने हे दाखवून दिले की स्थानिक राज्य इतके महत्त्वाचे का आहे—तुम्ही आधीच कोणती माहिती पाहिली आहे याचा मागोवा ठेवू देते. स्थानिक प्रवेश नेटवर्क विलंब टाळतो, अनुप्रयोग अधिक कार्यक्षम आणि त्रुटी-प्रतिरोधक बनवतो.

कोणतेही रोलअप किंवा एकत्रीकरण ऑपरेशन करत असताना, तुम्ही स्टेट स्टोअरचे नाव नमूद करणे आवश्यक आहे. रोलअप आणि एग्रीगेशन ऑपरेशन्स KTable उदाहरण देते, आणि KTable जुन्या परिणामांच्या जागी नवीन परिणाम देण्यासाठी स्टेट स्टोरेज वापरते. तुम्ही पाहिल्याप्रमाणे, सर्व अद्यतने पाइपलाइनवर पाठवली जात नाहीत आणि हे महत्त्वाचे आहे कारण एकत्रित ऑपरेशन्स सारांश माहिती तयार करण्यासाठी डिझाइन केलेले आहेत. तुम्ही स्थानिक स्थिती लागू न केल्यास, KTable सर्व एकत्रीकरण आणि रोलअप परिणाम फॉरवर्ड करेल.

पुढे, आम्ही ऑपरेशन्स जसे की विशिष्ट कालावधीत एकत्रित करणे - तथाकथित विंडोिंग ऑपरेशन्स पाहू.

५.३.२. विंडो ऑपरेशन्स

मागील विभागात, आम्ही स्लाइडिंग कॉन्व्होल्यूशन आणि एकत्रीकरण सादर केले. ऍप्लिकेशनने स्टॉक विक्री व्हॉल्यूमचे सतत रोल-अप केले, त्यानंतर एक्सचेंजवर सर्वाधिक व्यापार झालेल्या पाच समभागांचे एकत्रीकरण केले.

कधीकधी असे सतत एकत्रीकरण आणि परिणामांचे रोल-अप आवश्यक असते. आणि काहीवेळा आपल्याला केवळ दिलेल्या कालावधीत ऑपरेशन्स करण्याची आवश्यकता असते. उदाहरणार्थ, गेल्या 10 मिनिटांत विशिष्ट कंपनीच्या शेअर्ससह किती एक्सचेंज व्यवहार केले गेले याची गणना करा. किंवा गेल्या 15 मिनिटांत नवीन जाहिरात बॅनरवर किती वापरकर्त्यांनी क्लिक केले. एखादे ॲप्लिकेशन असे ऑपरेशन अनेक वेळा करू शकते, परंतु केवळ ठराविक कालावधीसाठी लागू होणार्‍या परिणामांसह (टाइम विंडो).

खरेदीदाराद्वारे विनिमय व्यवहार मोजणे

पुढील उदाहरणामध्ये, आम्ही अनेक व्यापार्‍यांमधील स्टॉक व्यवहारांचा मागोवा घेऊ—एकतर मोठ्या संस्था किंवा स्मार्ट वैयक्तिक फायनान्सर.

या ट्रॅकिंगची दोन संभाव्य कारणे आहेत. त्यापैकी एक म्हणजे बाजारातील नेते काय खरेदी/विक्री करत आहेत हे जाणून घेणे आवश्यक आहे. जर हे मोठे खेळाडू आणि अत्याधुनिक गुंतवणूकदारांना संधी दिसली, तर त्यांच्या धोरणाचे अनुसरण करण्यात अर्थ आहे. दुसरे कारण म्हणजे बेकायदेशीर इनसाइडर ट्रेडिंगची कोणतीही संभाव्य चिन्हे शोधण्याची इच्छा. हे करण्यासाठी, तुम्हाला महत्त्वाच्या प्रेस रीलिझसह मोठ्या विक्रीच्या स्पाइक्सच्या परस्परसंबंधाचे विश्लेषण करावे लागेल.

अशा ट्रॅकिंगमध्ये खालील चरणांचा समावेश आहे:

• स्टॉक-व्यवहार विषयावरून वाचनासाठी एक प्रवाह तयार करणे;
• खरेदीदार आयडी आणि स्टॉक चिन्हानुसार येणार्‍या रेकॉर्डचे गटबद्ध करणे. groupBy पद्धतीला कॉल केल्याने KGroupedStream क्लासचे उदाहरण मिळते;
• KGroupedStream.windowedBy पद्धत टाइम विंडोपर्यंत मर्यादित डेटा प्रवाह परत करते, जे विंडो केलेल्या एकत्रीकरणास अनुमती देते. विंडो प्रकारावर अवलंबून, एकतर TimeWindowedKStream किंवा SessionWindowedKStream परत केला जातो;
• एकत्रीकरण ऑपरेशनसाठी व्यवहार संख्या. विंडो केलेला डेटा फ्लो ठरवतो की या मोजणीमध्ये विशिष्ट रेकॉर्ड विचारात घेतले जाते की नाही;
• विषयावर परिणाम लिहिणे किंवा विकासादरम्यान ते कन्सोलवर आउटपुट करणे.

या अनुप्रयोगाचे टोपोलॉजी सोपे आहे, परंतु त्याचे स्पष्ट चित्र उपयुक्त ठरेल. चला अंजीर वर एक नजर टाकूया. ५.११.

पुढे, आम्ही विंडो ऑपरेशन्सची कार्यक्षमता आणि संबंधित कोड पाहू.

विंडोचे प्रकार

काफ्का प्रवाहांमध्ये तीन प्रकारच्या खिडक्या आहेत:

• सत्रीय
• "टंबलिंग" (टंबलिंग);
• सरकणे/हॉपिंग.

कोणता निवडायचा हे तुमच्या व्यवसायाच्या गरजांवर अवलंबून आहे. टंबलिंग आणि जंपिंग विंडो वेळ-मर्यादित आहेत, तर सत्र विंडो वापरकर्त्याच्या क्रियाकलापांद्वारे मर्यादित आहेत—सत्राचा कालावधी केवळ वापरकर्ता किती सक्रिय आहे यावर अवलंबून असतो. लक्षात ठेवण्याची मुख्य गोष्ट अशी आहे की सर्व विंडोचे प्रकार नोंदींच्या तारखेवर/वेळेच्या शिक्क्यांवर आधारित असतात, प्रणालीच्या वेळेवर नाही.

पुढे, आम्ही प्रत्येक विंडो प्रकारासह आमचे टोपोलॉजी लागू करतो. संपूर्ण कोड फक्त पहिल्या उदाहरणात दिला जाईल; इतर प्रकारच्या विंडोसाठी विंडो ऑपरेशनच्या प्रकाराशिवाय काहीही बदलणार नाही.

सत्र विंडो

सेशन विंडो इतर सर्व प्रकारच्या खिडक्यांपेक्षा खूप वेगळ्या असतात. ते वापरकर्त्याच्या क्रियाकलापांनुसार (किंवा आपण ट्रॅक करू इच्छित असलेल्या घटकाच्या क्रियाकलापांनुसार) वेळेनुसार मर्यादित नाहीत. सत्र विंडो निष्क्रियतेच्या कालावधीनुसार मर्यादित केल्या जातात.

आकृती 5.12 सत्र विंडोची संकल्पना स्पष्ट करते. लहान सत्र त्याच्या डावीकडील सत्रात विलीन होईल. आणि उजवीकडील सत्र वेगळे असेल कारण ते निष्क्रियतेच्या दीर्घ कालावधीचे अनुसरण करते. सत्र विंडो वापरकर्त्याच्या क्रियाकलापावर आधारित असतात, परंतु प्रवेश कोणत्या सत्राशी संबंधित आहे हे निर्धारित करण्यासाठी प्रविष्ट्यांमधून तारीख/वेळ स्टॅम्प वापरतात.

स्टॉक व्यवहारांचा मागोवा घेण्यासाठी सत्र विंडो वापरणे

विनिमय व्यवहारांबद्दल माहिती मिळविण्यासाठी सत्र विंडो वापरू. सत्र विंडोची अंमलबजावणी सूची 5.5 मध्ये दर्शविली आहे (जी src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java मध्ये आढळू शकते).

तुम्ही या टोपोलॉजीमधील बहुतांश ऑपरेशन्स आधीच पाहिल्या आहेत, त्यामुळे त्यांना येथे पुन्हा पाहण्याची गरज नाही. परंतु येथे अनेक नवीन घटक देखील आहेत, ज्यांची आपण आता चर्चा करू.

कोणतेही ग्रुपबाय ऑपरेशन विशेषत: काही प्रकारचे एकत्रीकरण ऑपरेशन (एकत्रीकरण, रोलअप किंवा मोजणी) करते. तुम्ही एकतर चालू एकूण एकत्रीकरण किंवा विंडो एकत्रीकरण करू शकता, जे एका निर्दिष्ट वेळेच्या विंडोमध्ये खाते रेकॉर्ड घेते.

सूची 5.5 मधील कोड सत्र विंडोमधील व्यवहारांची संख्या मोजतो. अंजीर मध्ये. 5.13 या क्रियांचे टप्प्याटप्प्याने विश्लेषण केले जाते.

windowedBy(SessionWindows.with(twentySeconds).until(20Minutes)) वर कॉल करून आम्ही 15 सेकंदांच्या निष्क्रियता अंतराल आणि 20 मिनिटांच्या सातत्य अंतरासह सत्र विंडो तयार करतो. 20 सेकंदांच्या निष्क्रिय मध्यांतराचा अर्थ असा आहे की चालू सत्राच्या समाप्तीच्या XNUMX सेकंदांच्या आत किंवा चालू (सक्रिय) सत्रामध्ये येणारी कोणतीही एंट्री ऍप्लिकेशनमध्ये समाविष्ट असेल.

पुढे, सत्र विंडोमध्ये कोणते एकत्रीकरण ऑपरेशन करणे आवश्यक आहे ते आम्ही निर्दिष्ट करतो - या प्रकरणात, गणना करा. इनकमिंग एंट्री निष्क्रियता विंडोच्या बाहेर पडल्यास (तारीख/वेळ स्टॅम्पच्या दोन्ही बाजूला), अनुप्रयोग नवीन सत्र तयार करतो. धारणा मध्यांतर म्हणजे ठराविक वेळेसाठी सत्र राखणे आणि सत्राच्या निष्क्रियतेच्या कालावधीच्या पलीकडे विस्तारलेल्या परंतु तरीही संलग्न केला जाऊ शकतो अशा उशीरा डेटासाठी परवानगी देतो. याव्यतिरिक्त, विलीनीकरणाच्या परिणामी नवीन सत्राची सुरुवात आणि समाप्ती सर्वात जुनी आणि नवीनतम तारीख/वेळ स्टॅम्पशी संबंधित आहे.

सत्रे कशी कार्य करतात हे पाहण्यासाठी मोजणी पद्धतीतील काही नोंदी पाहू (तक्ता 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 ची समाप्ती वेळ.

या विभागात वर्णन केलेल्या गोष्टींवरून, खालील महत्त्वपूर्ण बारकावे लक्षात ठेवण्यासारखे आहे:

• सत्रे निश्चित आकाराच्या विंडो नाहीत. सत्राचा कालावधी दिलेल्या कालावधीतील क्रियाकलापांद्वारे निर्धारित केला जातो;
• डेटामधील तारीख/वेळ स्टॅम्प निर्धारित करतात की इव्हेंट विद्यमान सत्रामध्ये आहे की निष्क्रिय कालावधी दरम्यान.

पुढे आपण पुढील प्रकारच्या विंडो - “टंबलिंग” विंडोबद्दल चर्चा करू.

"टंबलिंग" खिडक्या

टंबलिंग विंडो ठराविक कालावधीत घडणाऱ्या घटना कॅप्चर करतात. अशी कल्पना करा की तुम्हाला दर 20 सेकंदांनी एका विशिष्ट कंपनीचे सर्व स्टॉक व्यवहार कॅप्चर करणे आवश्यक आहे, त्यामुळे तुम्ही त्या कालावधीतील सर्व घटना गोळा कराल. 20-सेकंदांच्या मध्यांतराच्या शेवटी, विंडो फिरते आणि नवीन 20-सेकंद निरीक्षण मध्यांतराकडे जाते. आकृती 5.14 ही परिस्थिती स्पष्ट करते.

जसे आपण पाहू शकता, शेवटच्या 20 सेकंदात प्राप्त झालेल्या सर्व घटना विंडोमध्ये समाविष्ट केल्या आहेत. या कालावधीच्या शेवटी, एक नवीन विंडो तयार केली जाते.

सूची 5.6 कोड दाखवते जे दर 20 सेकंदांनी स्टॉक व्यवहार कॅप्चर करण्यासाठी टंबलिंग विंडोचा वापर दर्शवते (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java मध्ये आढळते).

TimeWindows.of मेथड कॉलमध्ये या छोट्या बदलासह, तुम्ही टंबलिंग विंडो वापरू शकता. हे उदाहरण till() पद्धतीला कॉल करत नाही, म्हणून 24 तासांचा डीफॉल्ट धारणा अंतराल वापरला जाईल.

शेवटी, विंडो पर्यायांच्या शेवटच्या - "हॉपिंग" विंडोकडे जाण्याची वेळ आली आहे.

स्लाइडिंग ("उडी मारणे") खिडक्या

स्लाइडिंग/हॉपिंग विंडो टंबलिंग विंडो सारख्याच असतात, परंतु थोड्या फरकाने. अलीकडील इव्हेंटवर प्रक्रिया करण्यासाठी नवीन विंडो तयार करण्यापूर्वी स्लाइडिंग विंडो वेळ मध्यांतर संपेपर्यंत प्रतीक्षा करत नाहीत. ते विंडो कालावधीपेक्षा कमी प्रतीक्षा अंतरानंतर नवीन गणना सुरू करतात.

टंबलिंग आणि जंपिंग विंडोमधील फरक स्पष्ट करण्यासाठी, स्टॉक एक्सचेंज व्यवहार मोजण्याच्या उदाहरणाकडे परत जाऊ या. आमचे ध्येय अजूनही व्यवहारांची संख्या मोजण्याचे आहे, परंतु आम्ही काउंटर अपडेट करण्यापूर्वी संपूर्ण वेळ प्रतीक्षा करू इच्छित नाही. त्याऐवजी, आम्ही कमी अंतराने काउंटर अपडेट करू. उदाहरणार्थ, आम्ही तरीही दर 20 सेकंदांनी व्यवहारांची संख्या मोजू, परंतु अंजीर मध्ये दर्शविल्याप्रमाणे, दर 5 सेकंदांनी काउंटर अद्यतनित करू. ५.१५. या प्रकरणात, आम्ही ओव्हरलॅपिंग डेटासह तीन परिणाम विंडोसह समाप्त करतो.

सूची 5.7 स्लाइडिंग विंडो परिभाषित करण्यासाठी कोड दर्शवते (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java मध्ये आढळते).

AdvanceBy() पद्धतीत कॉल जोडून टंबलिंग विंडोला हॉपिंग विंडोमध्ये रूपांतरित केले जाऊ शकते. दर्शविलेल्या उदाहरणामध्ये, बचत मध्यांतर 15 मिनिटे आहे.

एकत्रीकरण परिणाम वेळ विंडोमध्ये कसे मर्यादित करायचे ते तुम्ही या विभागात पाहिले. विशेषतः, या विभागातील खालील तीन गोष्टी तुम्ही लक्षात ठेवाव्यात अशी माझी इच्छा आहे:

• सत्र विंडोचा आकार कालावधीनुसार मर्यादित नाही, परंतु वापरकर्त्याच्या क्रियाकलापांद्वारे मर्यादित आहे;
• "टंबलिंग" विंडो दिलेल्या कालावधीतील घटनांचे विहंगावलोकन प्रदान करते;
• जंपिंग विंडोचा कालावधी निश्चित आहे, परंतु ते वारंवार अपडेट केले जातात आणि सर्व विंडोमध्ये ओव्हरलॅपिंग एंट्री असू शकतात.

पुढे, कनेक्शनसाठी KTable परत KStream मध्ये कसे रूपांतरित करायचे ते आपण शिकू.

५.३.३. 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 मेथडला कॉल करून, इंडस्ट्री नावाने की बदला, आणि नंतर Windowed उदाहरणातून 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).

हे लक्षात घेण्यासारखे आहे की कोणत्याही Serde ऑब्जेक्ट्स निर्दिष्ट करणे आवश्यक नाही, कारण स्ट्रिंग Serdes सेटिंग्जमध्ये वापरली जाते. तसेच, EARLIEST गणनेचा वापर करून, सारणी अगदी सुरुवातीलाच नोंदींनी भरलेली असते.

आता आपण अंतिम टप्प्यावर जाऊ शकतो - कनेक्शन.

व्यवहार गणना डेटासह बातम्या अद्यतने कनेक्ट करणे

कनेक्शन तयार करणे कठीण नाही. संबंधित उद्योगासाठी स्टॉक न्यूज नसल्यास आम्ही लेफ्ट जॉईन वापरू (आवश्यक कोड src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java फाइलमध्ये आढळू शकतो) (लिस्टिंग 5.10).

हा leftJoin ऑपरेटर अगदी सोपा आहे. Chapter 4 मधील जोडण्यांप्रमाणे, JoinWindow पद्धत वापरली जात नाही कारण KStream-KTable जॉईन करताना, प्रत्येक कीसाठी KTable मध्ये फक्त एकच एंट्री असते. असे कनेक्शन वेळेत मर्यादित नाही: रेकॉर्ड एकतर केटेबलमध्ये आहे किंवा अनुपस्थित आहे. मुख्य निष्कर्ष: KTable ऑब्जेक्ट्स वापरून तुम्ही कमी वारंवार अद्यतनित केलेल्या संदर्भ डेटासह KStream समृद्ध करू शकता.

आता आपण KStream मधील इव्हेंट समृद्ध करण्याचा अधिक कार्यक्षम मार्ग पाहू.

५.३.४. ग्लोबलकेटेबल ऑब्जेक्ट्स

तुम्ही बघू शकता, इव्हेंट प्रवाह समृद्ध करण्याची किंवा त्यात संदर्भ जोडण्याची गरज आहे. धडा 4 मध्ये तुम्ही दोन KStream ऑब्जेक्ट्समधील कनेक्शन पाहिले, आणि मागील विभागात तुम्ही KStream आणि KTable मधील कनेक्शन पाहिले. या सर्व प्रकरणांमध्ये, नवीन प्रकार किंवा मूल्यासाठी की मॅप करताना डेटा प्रवाहाचे पुन्हा विभाजन करणे आवश्यक आहे. कधीकधी पुनर्विभाजन स्पष्टपणे केले जाते, आणि काहीवेळा काफ्का प्रवाह ते आपोआप करते. री-विभाजन आवश्यक आहे कारण की बदलल्या आहेत आणि रेकॉर्ड नवीन विभागांमध्ये संपल्या पाहिजेत, अन्यथा कनेक्शन अशक्य होईल (याची चर्चा धडा 4 मध्ये, उपविभाग 4.2.4 मधील “डेटा पुन्हा-विभाजन” मध्ये केली आहे).

पुनर्विभाजनाची किंमत आहे

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

लहान डेटासेटशी कनेक्ट करत आहे

काही प्रकरणांमध्ये, कनेक्ट करण्यासाठी संदर्भ डेटाची मात्रा तुलनेने लहान असते, म्हणून त्याच्या संपूर्ण प्रती प्रत्येक नोडवर स्थानिक पातळीवर सहजपणे बसू शकतात. अशा परिस्थितीसाठी, काफ्का स्ट्रीम्स ग्लोबलकेटेबल क्लास प्रदान करते.

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

केस्ट्रीम ऑब्जेक्ट्सला ग्लोबलकेटेबल ऑब्जेक्ट्सशी जोडत आहे

उपविभाग 5.3.2 मध्ये, आम्ही खरेदीदारांद्वारे विनिमय व्यवहारांचे विंडो एकत्रीकरण केले. या एकत्रीकरणाचे परिणाम असे काहीतरी दिसले:

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

या निकालांनी उद्देश पूर्ण केला असता, ग्राहकाचे नाव आणि कंपनीचे संपूर्ण नाव देखील प्रदर्शित केले असते तर ते अधिक उपयुक्त ठरले असते. ग्राहकाचे नाव आणि कंपनीचे नाव जोडण्यासाठी, तुम्ही सामान्य जोडणी करू शकता, परंतु तुम्हाला दोन की मॅपिंग आणि पुन्हा विभाजन करणे आवश्यक आहे. GlobalKTable सह तुम्ही अशा ऑपरेशन्सची किंमत टाळू शकता.

हे करण्यासाठी, आम्ही सूची 5.11 मधील काउंटस्ट्रीम ऑब्जेक्ट वापरू (संबंधित कोड src/main/java/bbejeck/chapter_5/GlobalKTableExample.java मध्ये आढळू शकतो) आणि त्यास दोन GlobalKTable ऑब्जेक्ट्सशी जोडू.

आम्ही यापूर्वीच याबद्दल चर्चा केली आहे, म्हणून मी त्याची पुनरावृत्ती करणार नाही. पण मी लक्षात घेतो की toStream().map फंक्शनमधील कोड वाचनीयतेसाठी इनलाइन लॅम्बडा एक्सप्रेशनऐवजी फंक्शन ऑब्जेक्टमध्ये अॅब्स्ट्रॅक्ट केला जातो.

पुढील पायरी म्हणजे 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 नुसार कनेक्टिव्हिटी क्षमता प्रतिबिंबित करते; भविष्यातील प्रकाशनांमध्ये काहीतरी बदलू शकते.

गोष्टी गुंडाळण्यासाठी, मूलभूत गोष्टींचा आढावा घेऊया: तुम्ही स्थानिक स्थिती वापरून इव्हेंट स्ट्रीम (KStream) आणि अपडेट स्ट्रीम (KTable) कनेक्ट करू शकता. वैकल्पिकरित्या, संदर्भ डेटाचा आकार खूप मोठा नसल्यास, तुम्ही GlobalKTable ऑब्जेक्ट वापरू शकता. ग्लोबलकेटेबल्स प्रत्येक काफ्का स्ट्रीम्स ऍप्लिकेशन नोडवर सर्व विभाजनांची प्रतिकृती बनवतात, की कोणत्या विभाजनाशी संबंधित आहे याची पर्वा न करता सर्व डेटा उपलब्ध असल्याचे सुनिश्चित करते.

पुढे आपण काफ्का स्ट्रीम्स वैशिष्ट्य पाहू, ज्यामुळे आपण काफ्का विषयातील डेटा वापरल्याशिवाय स्थितीतील बदल पाहू शकतो.

५.३.५. प्रश्नार्थक स्थिती

आम्ही आधीच अनेक ऑपरेशन केले आहेत ज्यामध्ये राज्य समाविष्ट आहे आणि नेहमी कन्सोलवर परिणाम आउटपुट करतो (विकासाच्या हेतूंसाठी) किंवा त्यांना एखाद्या विषयावर (उत्पादन हेतूंसाठी) लिहा. एखाद्या विषयावर परिणाम लिहिताना, ते पाहण्यासाठी तुम्हाला काफ्का ग्राहक वापरावा लागेल.

या विषयांवरील डेटा वाचणे हे एक प्रकारचे भौतिक दृश्य मानले जाऊ शकते. आमच्या उद्देशांसाठी, आम्ही विकिपीडिया मधील भौतिक दृश्याची व्याख्या वापरू शकतो: “...एक भौतिक डेटाबेस ऑब्जेक्ट ज्यामध्ये क्वेरीचे परिणाम असतात. उदाहरणार्थ, ती रिमोट डेटाची स्थानिक प्रत, किंवा सारणीच्या पंक्ती आणि/किंवा स्तंभांचा उपसंच असू शकतो किंवा परिणाम सामील होऊ शकतो, किंवा एकत्रीकरणाद्वारे प्राप्त केलेला सारांश सारणी असू शकते" (https://en.wikipedia.org/wiki /Materialized_view).

काफ्का स्ट्रीम्स तुम्हाला स्टेट स्टोअर्सवर इंटरएक्टिव्ह क्वेरी चालवण्याची अनुमती देते, ज्यामुळे तुम्हाला ही भौतिक दृश्ये थेट वाचता येतात. हे लक्षात घेणे महत्त्वाचे आहे की स्टेट स्टोअरची क्वेरी ही केवळ-वाचनीय ऑपरेशन आहे. हे सुनिश्चित करते की तुमचा ॲप्लिकेशन डेटावर प्रक्रिया करत असताना तुम्हाला चुकूनही स्थिती विसंगत बनवण्याची काळजी करण्याची गरज नाही.

राज्य स्टोअर्सची थेट क्वेरी करण्याची क्षमता महत्त्वाची आहे. याचा अर्थ असा की तुम्ही काफ्का ग्राहकाकडून प्रथम डेटा आणल्याशिवाय डॅशबोर्ड अॅप्लिकेशन्स तयार करू शकता. हे अनुप्रयोगाची कार्यक्षमता देखील वाढवते, कारण पुन्हा डेटा लिहिण्याची आवश्यकता नाही:

• डेटाच्या स्थानिकतेबद्दल धन्यवाद, ते द्रुतपणे ऍक्सेस केले जाऊ शकतात;
• डेटाचे डुप्लिकेशन काढून टाकले जाते, कारण ते बाह्य संचयनावर लिहिलेले नाही.

मुख्य गोष्ट तुम्ही लक्षात ठेवावी अशी माझी इच्छा आहे की तुम्ही तुमच्या अर्जामधून थेट स्थितीची चौकशी करू शकता. यामुळे तुम्हाला मिळणाऱ्या संधींचा अतिरेक करता येणार नाही. काफ्का कडील डेटा वापरण्याऐवजी आणि अनुप्रयोगासाठी डेटाबेसमध्ये रेकॉर्ड संग्रहित करण्याऐवजी, आपण समान परिणामांसह स्टेट स्टोअर्सची क्वेरी करू शकता. राज्य स्टोअरमध्ये थेट क्वेरी म्हणजे कमी कोड (ग्राहक नाही) आणि कमी सॉफ्टवेअर (परिणाम संचयित करण्यासाठी डेटाबेस टेबलची आवश्यकता नाही).

आम्ही या धड्यात थोडासा ग्राउंड कव्हर केला आहे, म्हणून आम्ही आत्तासाठी राज्य स्टोअर्स विरुद्धच्या परस्परसंवादी प्रश्नांची चर्चा सोडू. पण काळजी करू नका: धडा 9 मध्ये, आम्ही संवादात्मक प्रश्नांसह एक साधा डॅशबोर्ड अनुप्रयोग तयार करू. हे या आणि मागील प्रकरणांमधील काही उदाहरणे परस्परसंवादी क्वेरी आणि तुम्ही काफ्का स्ट्रीम्स ऍप्लिकेशन्समध्ये कसे जोडू शकता हे दाखवण्यासाठी वापरेल.

सारांश

• KStream ऑब्जेक्ट्स घटनांच्या प्रवाहाचे प्रतिनिधित्व करतात, डेटाबेसमधील इन्सर्टशी तुलना करता येतात. KTable ऑब्जेक्ट्स अपडेट स्ट्रीम्सचे प्रतिनिधित्व करतात, जसे की डेटाबेसमधील अपडेट्स. KTable ऑब्जेक्टचा आकार वाढत नाही, जुन्या नोंदी नवीन बदलल्या जातात.
• एकत्रीकरण ऑपरेशन्ससाठी KTable ऑब्जेक्ट्स आवश्यक आहेत.
• विंडोिंग ऑपरेशन्स वापरून, तुम्ही एकत्रित डेटा टाइम बकेटमध्ये विभाजित करू शकता.
• GlobalKTable ऑब्जेक्ट्सबद्दल धन्यवाद, तुम्ही विभाजनाची पर्वा न करता अनुप्रयोगात कुठेही संदर्भ डेटामध्ये प्रवेश करू शकता.
• KStream, KTable आणि GlobalKTable ऑब्जेक्ट्समधील कनेक्शन शक्य आहेत.

आतापर्यंत, आम्ही उच्च-स्तरीय KStream DSL वापरून काफ्का प्रवाह अनुप्रयोग तयार करण्यावर लक्ष केंद्रित केले आहे. जरी उच्च-स्तरीय दृष्टीकोन तुम्हाला नीटनेटके आणि संक्षिप्त कार्यक्रम तयार करण्यास अनुमती देतो, तरीही ते वापरणे व्यापार-बंद दर्शवते. DSL KStream सह कार्य करणे म्हणजे नियंत्रणाची डिग्री कमी करून तुमच्या कोडची संक्षिप्तता वाढवणे. पुढील प्रकरणामध्ये, आम्ही निम्न-स्तरीय हँडलर नोड API पाहू आणि इतर ट्रेड-ऑफ वापरून पाहू. प्रोग्राम्स पूर्वीपेक्षा जास्त लांब असतील, परंतु आम्हाला आवश्यक असणारे जवळपास कोणतेही हँडलर नोड तयार करण्यात आम्ही सक्षम होऊ.

→ पुस्तकाबद्दल अधिक तपशील येथे आढळू शकतात प्रकाशकाची वेबसाइट

→ Habrozhiteli साठी कूपन वापरून 25% सूट - काफ्का प्रवाह

→ पुस्तकाच्या कागदी आवृत्तीसाठी पैसे दिल्यावर, इलेक्ट्रॉनिक पुस्तक ई-मेलद्वारे पाठवले जाईल.

स्त्रोत: www.habr.com