पुस्तक "काफ्का स्ट्रिम्स इन एक्शन। वास्तविक-समय कामको लागि अनुप्रयोगहरू र माइक्रोसेवाहरू"

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

अंश। ५.३। एग्रीगेशन र विन्डोिङ अपरेसनहरू

यस खण्डमा, हामी काफ्का स्ट्रिमहरूको सबैभन्दा आशाजनक भागहरू अन्वेषण गर्न अगाडि बढ्नेछौं। अहिलेसम्म हामीले काफ्का स्ट्रिमका निम्न पक्षहरूलाई समेटेका छौं:

• एक प्रशोधन टोपोलजी सिर्जना;
• स्ट्रिमिङ अनुप्रयोगहरूमा राज्य प्रयोग गर्दै;
• डाटा स्ट्रिम जडानहरू प्रदर्शन गर्दै;
• घटना स्ट्रिमहरू (KStream) र अद्यावधिक स्ट्रिमहरू (KTable) बीचको भिन्नता।

निम्न उदाहरणहरूमा हामी यी सबै तत्वहरू सँगै ल्याउनेछौं। तपाईंले स्ट्रिमिङ एप्लिकेसनहरूको अर्को उत्कृष्ट सुविधा, windowing बारे पनि सिक्नुहुनेछ। हाम्रो पहिलो उदाहरण एक साधारण एकत्रीकरण हुनेछ।

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

स्ट्रिमिङ डेटा संग काम गर्दा एकत्रीकरण र समूह महत्वपूर्ण उपकरण हो। प्राप्त भएका व्यक्तिगत अभिलेखहरूको जाँच प्रायः अपर्याप्त हुन्छ। डाटाबाट थप जानकारी निकाल्न, यो समूह र तिनीहरूलाई संयोजन गर्न आवश्यक छ।

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

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

1. कच्चा स्टक ट्रेडिंग जानकारी प्रकाशित गर्ने विषय-आधारित स्रोत सिर्जना गर्नुहोस्। हामीले स्टक ट्रान्जेक्शन प्रकारको वस्तुलाई ShareVolume प्रकारको वस्तुमा नक्सा गर्नुपर्नेछ। बिन्दु यो हो कि स्टक लेनदेन वस्तुले बिक्री मेटाडेटा समावेश गर्दछ, तर हामीलाई बिक्री भइरहेको सेयरहरूको संख्याको बारेमा मात्र डाटा चाहिन्छ।
2. समूह शेयर भोल्युम डेटा स्टक प्रतीक द्वारा। एकपटक प्रतीकद्वारा समूहबद्ध भएपछि, तपाईंले यो डेटालाई स्टक बिक्री भोल्युमहरूको उप-योगमा संक्षिप्त गर्न सक्नुहुन्छ। यो ध्यान दिन लायक छ कि KStream.groupBy विधिले KGroupedStream प्रकारको उदाहरण फर्काउँछ। र तपाईले केडीईटेबल उदाहरण प्राप्त गर्न सक्नुहुन्छ केडीई ग्रुप स्ट्रिम.रिड्युस विधिलाई कल गरेर।

केडीई समूह स्ट्रिम इन्टरफेस के हो

KStream.groupBy र KStream.groupByKey विधिहरूले KGroupedStream को उदाहरण फर्काउँछ। KGroupedStream कुञ्जीहरूद्वारा समूहबद्ध गरेपछि घटनाहरूको स्ट्रिमको मध्यवर्ती प्रतिनिधित्व हो। यो सबैसँग प्रत्यक्ष कामको लागि अभिप्रेरित छैन। यसको सट्टा, केडीई समूह स्ट्रिम एग्रिगेसन सञ्चालनका लागि प्रयोग गरिन्छ, जसको परिणाम सधैं केडीई तालिका हुन्छ। र जब एकीकरण कार्यहरूको नतिजा 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

केडीई स्ट्रिम इन्टरफेससँग रेकर्डहरू समूहबद्ध गर्नका लागि दुई तरिकाहरू छन्: GroupByKey र GroupBy। दुवैले केडीई समूहबद्ध तालिका फर्काउँछन्, त्यसैले तपाइँ सोचिरहनुभएको हुन सक्छ कि तिनीहरू बीचको भिन्नता के हो र कुन कहिले प्रयोग गर्ने?

GroupByKey विधि प्रयोग गरिन्छ जब केडीई स्ट्रिममा कुञ्जीहरू पहिले नै खाली छैनन्। र सबैभन्दा महत्त्वपूर्ण कुरा, "पुनः विभाजन आवश्यक छ" झण्डा कहिल्यै सेट गरिएको थिएन।

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

mapValues ​​र groupBy विधिहरूले के गर्छ भन्ने कुरा स्पष्ट छ, त्यसैले sum() विधि (src/main/java/bbejeck/model/ShareVolume.java मा फेला परेको) (लिस्टिङ ५.३) मा हेरौं।

ShareVolume.sum विधिले स्टक बिक्री भोल्युमको चलिरहेको कुल फिर्ता गर्छ, र गणनाको सम्पूर्ण श्रृंखलाको परिणाम एक KTable वस्तु हो। । अब तपाईले केटेबलले खेलेको भूमिका बुझ्नुहुन्छ। जब ShareVolume वस्तुहरू आइपुग्छ, सम्बन्धित KTable वस्तुले हालको अद्यावधिक भण्डारण गर्छ। यो सम्झना महत्त्वपूर्ण छ कि सबै अद्यावधिकहरू अघिल्लो shareVolumeKTable मा प्रतिबिम्बित छन्, तर सबै थप पठाइएका छैनन्।

त्यसपछि हामी यो KTable को प्रयोग गरी प्रत्येक उद्योगमा सबैभन्दा धेरै सेयर कारोबार भएका पाँच कम्पनीहरूमा पुग्नको लागि (ट्रेड गरिएको सेयरको संख्या अनुसार) जम्मा गर्छौं। यस मामला मा हाम्रो कार्यहरु पहिलो एकत्रीकरण को लागी समान हुनेछ।

1. उद्योग द्वारा व्यक्तिगत शेयर भोल्युम वस्तुहरू समूह गर्न अर्को समूहद्वारा सञ्चालन गर्नुहोस्।
2. ShareVolume वस्तुहरू संक्षेप गर्न सुरु गर्नुहोस्। यस पटक एग्रीगेशन वस्तु निश्चित आकारको प्राथमिकता लाम हो। यो निश्चित आकारको लाइनमा, सबैभन्दा धेरै सेयर बिक्री भएका पाँच कम्पनीहरू मात्र राखिएका छन्।
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 ले पुरानो प्रविष्टिलाई नयाँसँग बदल्छ। एग्रीगेशन एक समान तरिकामा हुन्छ: एउटै कुञ्जी संग नवीनतम रेकर्ड एकत्रित छन्। जब एक रेकर्ड आउँछ, यो एक adder प्रयोग गरेर FixedSizePriorityQueue वर्ग उदाहरणमा थपिन्छ (समग्र विधि कलमा दोस्रो प्यारामिटर), तर यदि उही कुञ्जीसँग अर्को रेकर्ड पहिले नै अवस्थित छ भने, पुरानो रेकर्ड घटाउने प्रयोग गरेर हटाइन्छ (तेस्रो प्यारामिटर समग्र विधि कल)।

यसको मतलब यो हो कि हाम्रो एग्रीगेटर, FixedSizePriorityQueue ले सबै मानहरू एउटै कुञ्जीमा जम्मा गर्दैन, तर N सबैभन्दा धेरै ट्रेड गरिएका स्टकहरूको परिमाणहरूको गतिशील योग भण्डारण गर्छ। प्रत्येक आगमन प्रविष्टिले अहिलेसम्म बेचेको शेयरहरूको कुल संख्या समावेश गर्दछ। KTable ले तपाइँलाई जानकारी दिनेछ कि कुन कम्पनीको शेयर हाल सबैभन्दा बढी कारोबार भएको छ, प्रत्येक अपडेटको रोलिङ एग्रीगेसन आवश्यक नपर्ने।

हामीले दुईवटा महत्त्वपूर्ण कुराहरू गर्न सिक्यौं:

• साझा कुञ्जीद्वारा केडीई टेबलमा समूह मानहरू;
• यी समूहबद्ध मानहरूमा रोलअप र एग्रीगेसन जस्ता उपयोगी कार्यहरू प्रदर्शन गर्नुहोस्।

काफ्का स्ट्रिम्स एप्लिकेसन मार्फत सर्ने डाटाको अर्थ बुझ्न र यसले के जानकारी बोक्छ भन्ने बुझ्नको लागि यी कार्यहरू कसरी गर्ने भन्ने कुरा जान्न महत्त्वपूर्ण छ।

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

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

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

५.३.२। सञ्झ्याल सञ्चालन

अघिल्लो खण्डमा, हामीले स्लाइडिङ कन्भोलुसन र एग्रीगेसन प्रस्तुत गरेका थियौं। एप्लिकेसनले स्टक बिक्री भोल्युमको निरन्तर रोल-अप प्रदर्शन गर्‍यो, त्यसपछि एक्सचेन्जमा सबैभन्दा धेरै कारोबार भएका पाँच स्टकहरूको एकत्रीकरण।

कहिलेकाहीँ यस्तो निरन्तर एकत्रीकरण र परिणामहरूको रोल-अप आवश्यक हुन्छ। र कहिलेकाहीँ तपाईंले समयको एक निश्चित अवधिमा मात्र सञ्चालन गर्न आवश्यक छ। उदाहरणका लागि, पछिल्लो १० मिनेटमा कुनै विशेष कम्पनीको सेयरसँग कतिवटा विनिमय कारोबारहरू भएका छन् भनेर गणना गर्नुहोस्। वा पछिल्लो 10 मिनेटमा कति प्रयोगकर्ताहरूले नयाँ विज्ञापन ब्यानरमा क्लिक गरे। एप्लिकेसनले त्यस्ता कार्यहरू धेरै पटक गर्न सक्छ, तर परिणामहरू जुन निर्दिष्ट समय (समय सञ्झ्याल) मा मात्र लागू हुन्छ।

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

अर्को उदाहरणमा, हामी धेरै व्यापारीहरू - कि त ठूला संस्थाहरू वा स्मार्ट व्यक्तिगत फाइनान्सरहरू मार्फत स्टक लेनदेनहरू ट्र्याक गर्नेछौं।

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

यस्तो ट्र्याकिङ निम्न चरणहरू समावेश:

• स्टक-लेनदेन विषयबाट पढ्नको लागि स्ट्रिम सिर्जना गर्दै;
• क्रेता आईडी र स्टक प्रतीक द्वारा आगमन रेकर्ड समूहीकरण। समूहद्वारा विधि कल गर्दा केडीई समूह स्ट्रिम वर्गको उदाहरण फर्काउँछ;
• KGroupedStream.windowedBy विधिले समय सञ्झ्यालमा सीमित डाटा स्ट्रिम फर्काउँछ, जसले सञ्झ्याल एकत्रीकरणलाई अनुमति दिन्छ। सञ्झ्याल प्रकारको आधारमा, या त TimeWindowedKStream वा SessionWindowedKStream फर्काइन्छ;
• एकत्रीकरण कार्यको लागि लेनदेन गणना। विन्डो गरिएको डेटा प्रवाहले यो गणनामा एक विशेष रेकर्डलाई ध्यानमा राखिएको छ कि छैन भनेर निर्धारण गर्दछ;
• शीर्षकमा परिणामहरू लेख्ने वा विकासको क्रममा कन्सोलमा आउटपुट गर्ने।

यस अनुप्रयोगको टोपोलोजी सरल छ, तर यसको स्पष्ट तस्वीर उपयोगी हुनेछ। चित्रमा एक नजर राखौं। ५.११।

अर्को, हामी सञ्झ्याल सञ्चालनको कार्यक्षमता र सम्बन्धित कोड हेर्नेछौं।

विन्डो प्रकारहरू

काफ्का स्ट्रिमहरूमा तीन प्रकारका झ्यालहरू छन्:

• सत्रीय;
• "टम्बलिंग";
• स्लाइडिङ/हपिङ।

कुन छनौट गर्ने तपाईंको व्यवसाय आवश्यकताहरूमा निर्भर गर्दछ। टम्बलिङ र जम्पिङ सञ्झ्यालहरू समय-सीमित छन्, जबकि सत्र सञ्झ्यालहरू प्रयोगकर्ता गतिविधिद्वारा सीमित छन्—सत्र(हरू) को अवधि प्रयोगकर्ता कति सक्रिय छ भनेर मात्र निर्धारण गरिन्छ। याद गर्न मुख्य कुरा यो हो कि सबै सञ्झ्याल प्रकारहरू प्रविष्टिहरूको मिति/समय स्ट्याम्पहरूमा आधारित हुन्छन्, प्रणाली समय होइन।

अर्को, हामी प्रत्येक विन्डो प्रकारको साथ हाम्रो टोपोलोजी लागू गर्छौं। पूर्ण कोड पहिलो उदाहरणमा मात्र दिइनेछ; अन्य प्रकारका सञ्झ्यालहरूको लागि सञ्झ्याल सञ्चालनको प्रकार बाहेक केही परिवर्तन हुनेछैन।

सत्र विन्डोज

सत्र सञ्झ्यालहरू अन्य सबै प्रकारका सञ्झ्यालहरू भन्दा धेरै फरक छन्। तिनीहरू प्रयोगकर्ताको गतिविधि (वा तपाईंले ट्र्याक गर्न चाहानु भएको इकाईको गतिविधि) द्वारा समय द्वारा सीमित छैनन्। सत्र सञ्झ्यालहरू निष्क्रियताको अवधिद्वारा छुट्याइएको छ।

चित्र ५.१२ ले सत्र सञ्झ्यालको अवधारणालाई चित्रण गर्दछ। सानो सत्र यसको बायाँतिरको सत्रसँग मर्ज हुनेछ। र दायाँतिरको सत्र अलग हुनेछ किनभने यसले निष्क्रियताको लामो अवधि पछ्याउँछ। सत्र सञ्झ्यालहरू प्रयोगकर्ता गतिविधिमा आधारित हुन्छन्, तर प्रविष्टिहरूबाट मिति/समय स्ट्याम्पहरू प्रयोग गर्नुहोस् जुन प्रविष्टि कुन सत्रको हो भनेर निर्धारण गर्न।

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

एक्सचेन्ज लेनदेनको बारेमा जानकारी लिनको लागि सत्र सञ्झ्याल प्रयोग गरौं। सत्र सञ्झ्यालको कार्यान्वयन सूची 5.5 मा देखाइएको छ (जुन src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java मा फेला पार्न सकिन्छ)।

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

कुनै पनि GroupBy अपरेशनले सामान्यतया केहि प्रकारको एग्रीगेसन अपरेशन (एकत्रीकरण, रोलअप, वा गणना) गर्दछ। तपाइँ चलिरहेको कुलको साथ या त संचयी एकत्रीकरण गर्न सक्नुहुन्छ, वा विन्डो एग्रीगेसन, जसले निर्दिष्ट समय विन्डो भित्र खाता रेकर्डहरूमा लिन्छ।

लिस्टिङ 5.5 मा भएको कोडले सत्र विन्डोज भित्रको लेनदेनको संख्या गणना गर्दछ। चित्रमा। 5.13 यी कार्यहरू चरणबद्ध रूपमा विश्लेषण गरिन्छ।

windowedBy(SessionWindows.with(twentySeconds).until(FifteenMinutes)) लाई कल गरेर हामी 20 सेकेन्डको निष्क्रियता अन्तराल र 15 मिनेटको निरन्तरता अन्तराल भएको सत्र सञ्झ्याल सिर्जना गर्छौं। 20 सेकेन्डको निष्क्रिय अन्तरालको अर्थ हो कि अनुप्रयोगले हालको (सक्रिय) सत्रमा हालको सत्रको अन्त्य वा सुरु भएको २० सेकेन्ड भित्र आइपुगेको कुनै पनि प्रविष्टि समावेश गर्दछ।

अर्को, हामीले सत्र सञ्झ्यालमा कुन एग्रीगेशन अपरेशन गर्न आवश्यक छ निर्दिष्ट गर्दछौं - यस अवस्थामा, गणना गर्नुहोस्। यदि आगमन प्रविष्टि निष्क्रियता विन्डो (मिति/समय स्ट्याम्पको कुनै पनि छेउमा) बाहिर आउँछ भने, अनुप्रयोगले नयाँ सत्र सिर्जना गर्दछ। रिटेन्सन अन्तराल भनेको निश्चित समयको लागि सत्र कायम राख्नु हो र सत्रको निष्क्रियता अवधि भन्दा बाहिर विस्तारित तर अझै पनि संलग्न गर्न सकिने ढिलो डाटाको लागि अनुमति दिन्छ। थप रूपमा, मर्जको परिणामस्वरूप नयाँ सत्रको सुरु र अन्त्य प्रारम्भिक र पछिल्लो मिति/समय स्ट्याम्पसँग मेल खान्छ।

सत्रहरूले कसरी काम गर्छ भनेर हेर्नको लागि गणना विधिबाट केही प्रविष्टिहरू हेरौं (तालिका ५.१)।

जब रेकर्डहरू आइपुग्छन्, हामी एउटै कुञ्जीसँग अवस्थित सत्रहरू खोज्छौं, हालको मिति/समय स्ट्याम्प - निष्क्रियता अन्तराल भन्दा कम समाप्ति समय, र हालको मिति/समय स्ट्याम्प + निष्क्रियता अन्तराल भन्दा ठूलो सुरु समय। यसलाई ध्यानमा राख्दै, तालिकाबाट चार प्रविष्टिहरू। 5.1 लाई निम्नानुसार एकल सत्रमा मर्ज गरिएको छ।

1. रेकर्ड 1 पहिले आइपुग्छ, त्यसैले सुरु समय अन्त समय बराबर छ र 00:00:00 छ।

2. अर्को, प्रविष्टि 2 आउँछ, र हामी सत्रहरू खोज्छौं जुन 23:59:55 भन्दा पहिले समाप्त हुँदैन र 00:00:35 भन्दा पछि सुरु हुँदैन। हामीले रेकर्ड 1 फेला पार्छौं र सत्र 1 र 2 लाई जोड्छौं। हामी सत्र 1 (पहिले) र सत्र 2 को अन्त्य समय (पछि) लिन्छौं, ताकि हाम्रो नयाँ सत्र 00:00:00 मा सुरु हुन्छ र 00 मा समाप्त हुन्छ: ००:१५।

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 ले यो अवस्थालाई चित्रण गर्दछ।

तपाईले देख्न सक्नुहुन्छ, पछिल्लो २० सेकेन्डमा प्राप्त भएका सबै घटनाहरू विन्डोमा समावेश छन्। समयको यस अवधिको अन्त्यमा, नयाँ विन्डो सिर्जना गरिएको छ।

सूची ५.६ ले कोड देखाउँछ जसले प्रत्येक २० सेकेन्डमा स्टक लेनदेनहरू क्याप्चर गर्न टम्बलिंग विन्डोको प्रयोग देखाउँछ (src/main/java/bbejeck/chapter_5.6/CountingWindowingAndKtableJoinExample.java मा फेला परेको छ)।

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

अन्तमा, यो सञ्झ्याल विकल्पहरूको अन्तिममा जाने समय हो - "हपिङ" विन्डोज।

स्लाइडिङ ("जम्पिङ") विन्डोज

स्लाइडिङ/हपिङ विन्डोहरू टम्बलिंग विन्डोज जस्तै हुन्छन्, तर थोरै भिन्नताका साथ। हालका घटनाहरू प्रशोधन गर्न नयाँ सञ्झ्याल सिर्जना गर्नु अघि स्लाइडिङ विन्डोहरूले समय अन्तरालको अन्त्यसम्म पर्खदैन। तिनीहरू विन्डो अवधि भन्दा कम प्रतीक्षा अन्तराल पछि नयाँ गणनाहरू सुरु गर्छन्।

टम्बलिंग र जम्पिङ विन्डोहरू बीचको भिन्नताहरू चित्रण गर्न, स्टक एक्सचेन्ज लेनदेनको गणना गर्ने उदाहरणमा फर्कौं। हाम्रो लक्ष्य अझै पनि लेनदेनको संख्या गणना गर्नु हो, तर हामी काउन्टर अद्यावधिक गर्नु अघि सम्पूर्ण समय पर्खन चाहँदैनौं। बरु, हामी छोटो अन्तरालहरूमा काउन्टर अपडेट गर्नेछौं। उदाहरणका लागि, हामी अझै पनि प्रत्येक 20 सेकेन्डमा लेनदेनको संख्या गणना गर्नेछौं, तर चित्रमा देखाइए अनुसार, प्रत्येक 5 सेकेन्डमा काउन्टर अपडेट गर्नेछौं। ५.१५ यस अवस्थामा, हामी ओभरल्यापिङ डेटाको साथ तीन परिणाम विन्डोहरूसँग समाप्त हुन्छौं।

सूची 5.7 ले स्लाइडिङ सञ्झ्याल परिभाषित गर्नको लागि कोड देखाउँछ (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java मा पाइन्छ)।

AdvanceBy() विधिमा कल थपेर टम्बलिंग विन्डोलाई हपिङ विन्डोमा रूपान्तरण गर्न सकिन्छ। देखाइएको उदाहरणमा, बचत अन्तराल 15 मिनेट हो।

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

• सत्र सञ्झ्यालको आकार समय अवधि द्वारा सीमित छैन, तर प्रयोगकर्ता गतिविधि द्वारा;
• "टम्बलिंग" सञ्झ्यालले दिइएको अवधि भित्रका घटनाहरूको सिंहावलोकन प्रदान गर्दछ;
• जम्पिङ सञ्झ्यालहरूको अवधि निश्चित छ, तर तिनीहरू बारम्बार अद्यावधिक हुन्छन् र सबै विन्डोहरूमा ओभरल्यापिङ प्रविष्टिहरू समावेश हुन सक्छन्।

अर्को, हामी जडानको लागि KTable लाई केडीई स्ट्रिममा रूपान्तरण गर्ने तरिका सिक्ने छौँ।

५.३.३। KStream र KTable वस्तुहरू जडान गर्दै

अध्याय 4 मा, हामीले दुई केडीई स्ट्रिम वस्तुहरू जडान गर्न छलफल गर्यौं। अब हामीले KTable र 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 वस्तु सिर्जना गर्न कोडको एउटा मात्र लाइन लाग्छ (कोड src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java मा फेला पार्न सकिन्छ) (लिस्टिङ ५.९)।

यो नोट गर्न लायक छ कि कुनै Serde वस्तुहरू निर्दिष्ट गर्न आवश्यक छैन, किनकि स्ट्रिङ Serdes सेटिङहरूमा प्रयोग गरिन्छ। साथै, प्रारम्भिक गणना प्रयोग गरेर, तालिका सुरुमै रेकर्डहरूले भरिएको छ।

अब हामी अन्तिम चरणमा जान सक्छौं - जडान।

लेनदेन गणना डेटा संग समाचार अद्यावधिक जडान

जडान बनाउन गाह्रो छैन। सान्दर्भिक उद्योगका लागि कुनै स्टक समाचार नभएको अवस्थामा हामी बायाँ जोड्ने प्रयोग गर्नेछौं (आवश्यक कोड फाइल src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (लिस्टिङ ५.१०) मा फेला पार्न सकिन्छ।

यो leftJoin अपरेटर एकदम सरल छ। अध्याय 4 मा जोडिएको जस्तो नभई, JoinWindow विधि प्रयोग गरिएको छैन किनभने KStream-KTable जोड्ने कार्य गर्दा, प्रत्येक कुञ्जीको लागि KTable मा एउटा मात्र प्रविष्टि हुन्छ। यस्तो जडान समय मा सीमित छैन: रेकर्ड या त केटेबल मा छ वा अनुपस्थित छ। मुख्य निष्कर्ष: KTable वस्तुहरू प्रयोग गरेर तपाईँले कम बारम्बार अद्यावधिक गरिएको सन्दर्भ डाटासँग केडीई स्ट्रिमलाई समृद्ध बनाउन सक्नुहुन्छ।

अब हामी केडीई स्ट्रिमबाट घटनाहरू समृद्ध गर्न अझ प्रभावकारी तरिका हेर्नेछौं।

५.३.४। GlobalKTable वस्तुहरू

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

पुन: विभाजन लागत छ

पुन: विभाजन गर्न लागतहरू चाहिन्छ - मध्यवर्ती विषयहरू सिर्जना गर्न, अर्को शीर्षकमा डुप्लिकेट डाटा भण्डारण गर्नको लागि अतिरिक्त स्रोत लागत; यसको अर्थ यस विषयबाट लेख्ने र पढ्नको कारणले बढेको विलम्बता पनि हो। थप रूपमा, यदि तपाइँ एक भन्दा बढी पक्ष वा आयामहरूमा सामेल हुन आवश्यक छ भने, तपाइँले जोडिहरूलाई चेन गर्नुपर्छ, नयाँ कुञ्जीहरूसँग रेकर्डहरू नक्सा गर्नुहोस्, र पुन: विभाजन प्रक्रिया पुन: चलाउनुहोस्।

साना डेटासेटहरूमा जडान गर्दै

केही अवस्थामा, जडान गर्न सन्दर्भ डेटाको मात्रा अपेक्षाकृत सानो छ, त्यसैले यसको पूर्ण प्रतिलिपिहरू सजिलै प्रत्येक नोडमा स्थानीय रूपमा फिट हुन सक्छ। यस्तो अवस्थाको लागि, काफ्का स्ट्रिम्सले GlobalKTable वर्ग प्रदान गर्दछ।

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 प्रकार्यमा रहेको कोड पढ्न योग्यताको लागि इनलाइन lambda अभिव्यक्तिको सट्टा प्रकार्य वस्तुमा सारिएको छ।

अर्को चरण भनेको GlobalKTable को दुई उदाहरणहरू घोषणा गर्नु हो (देखाइएको कोड src/main/java/bbejeck/chapter_5/GlobalKTableExample.java फाइलमा फेला पार्न सकिन्छ) (लिस्टिङ ५.१२)।

कृपया ध्यान दिनुहोस् कि विषय नामहरू गणना गरिएका प्रकारहरू प्रयोग गरी वर्णन गरिएको छ।

अब हामीसँग सबै कम्पोनेन्टहरू तयार छन्, जडानको लागि कोड लेख्न बाँकी छ (जुन फाइल src/main/java/bbejeck/chapter_5/GlobalKTableExample.java मा फेला पार्न सकिन्छ) (सूची 5.13)।

यद्यपि यस कोडमा दुईवटा जोडहरू छन्, तिनीहरू जङ्गलमा बाँधिएका छन् किनभने तिनीहरूको नतिजा छुट्टै प्रयोग गरिएको छैन। परिणामहरू सम्पूर्ण कार्यको अन्त्यमा प्रदर्शित हुन्छन्।

जब तपाइँ माथिको सम्मिलित अपरेसन चलाउनुहुन्छ, तपाइँ निम्न जस्तै परिणामहरू प्राप्त गर्नुहुनेछ:

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

सार परिवर्तन भएको छैन, तर यी परिणामहरू अझ स्पष्ट देखिन्छन्।

यदि तपाईंले अध्याय 4 मा काउन्ट डाउन गर्नुभयो भने, तपाईंले पहिले नै कार्यमा धेरै प्रकारका जडानहरू देख्नुभएको छ। तिनीहरू तालिकामा सूचीबद्ध छन्। ५.२। यो तालिकाले काफ्का स्ट्रिमहरूको संस्करण १.०.० को रूपमा जडान क्षमताहरू प्रतिबिम्बित गर्दछ; भविष्यका रिलीजहरूमा केही परिवर्तन हुन सक्छ।

चीजहरू समेट्नको लागि, आधारभूत कुराहरू पुन: क्याप गरौं: तपाईं घटना स्ट्रिमहरू (KStream) जडान गर्न सक्नुहुन्छ र स्थानीय अवस्था प्रयोग गरेर स्ट्रिमहरू (KTable) अद्यावधिक गर्न सक्नुहुन्छ। वैकल्पिक रूपमा, यदि सन्दर्भ डेटाको आकार धेरै ठूलो छैन भने, तपाईँले GlobalKTable वस्तु प्रयोग गर्न सक्नुहुन्छ। GlobalKTables ले प्रत्येक काफ्का स्ट्रिम्स एप्लिकेसन नोडमा सबै विभाजनहरू नक्कल गर्दछ, यो सुनिश्चित गर्दै कि कुन कुन विभाजन कुञ्जीसँग मेल खान्छ, सबै डाटा उपलब्ध छन्।

अर्को हामी काफ्का स्ट्रिम्स सुविधा देख्नेछौं, जसको लागि हामी काफ्का विषयबाट डाटा उपभोग नगरी राज्य परिवर्तनहरू अवलोकन गर्न सक्छौं।

५.३.५। प्रश्नयोग्य अवस्था

हामीले पहिले नै राज्य समावेश गरी धेरै अपरेसनहरू प्रदर्शन गरिसकेका छौं र सँधै परिणामहरू कन्सोलमा आउटपुट (विकास उद्देश्यका लागि) वा तिनीहरूलाई शीर्षकमा लेख्नुहोस् (उत्पादन उद्देश्यका लागि)। कुनै विषयमा नतिजा लेख्दा, तपाईंले तिनीहरूलाई हेर्न काफ्का उपभोक्ता प्रयोग गर्नुपर्छ।

यी विषयहरूबाट डेटा पढ्नुलाई भौतिक दृश्यहरूको प्रकार मान्न सकिन्छ। हाम्रो उद्देश्यका लागि, हामी विकिपिडियाबाट भौतिक दृश्यको परिभाषा प्रयोग गर्न सक्छौं: “...एक भौतिक डाटाबेस वस्तु जसमा क्वेरीको नतिजा हुन्छ। उदाहरणका लागि, यो टाढाको डेटाको स्थानीय प्रतिलिपि, वा तालिकाको पङ्क्तिहरू र/वा स्तम्भहरूको एक उपसेट वा परिणामहरू समावेश गर्न, वा एकीकरण मार्फत प्राप्त सारांश तालिका हुन सक्छ" (https://en.wikipedia.org/wiki /Materialized_view)।

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

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

• डाटा को स्थानीयता को लागी धन्यवाद, तिनीहरू चाँडै पहुँच गर्न सकिन्छ;
• डाटाको नक्कल हटाइएको छ, किनकि यो बाह्य भण्डारणमा लेखिएको छैन।

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

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

सारांश

• केडीई स्ट्रिम वस्तुहरूले घटनाहरूको स्ट्रिमहरू प्रतिनिधित्व गर्दछ, डाटाबेसमा सम्मिलितहरूसँग तुलना गर्न सकिन्छ। KTable वस्तुहरूले अपडेट स्ट्रिमहरू प्रतिनिधित्व गर्दछ, जस्तै डाटाबेसमा अद्यावधिकहरू। KTable वस्तुको आकार बढ्दैन, पुराना रेकर्डहरू नयाँहरूद्वारा प्रतिस्थापन गरिन्छ।
• एग्रीगेशन सञ्चालनका लागि केडीई टेबल वस्तुहरू आवश्यक छन्।
• सञ्झ्याल अपरेसनहरू प्रयोग गरेर, तपाईँले समय बाल्टीहरूमा एकत्रित डाटा विभाजन गर्न सक्नुहुन्छ।
• GlobalKTable वस्तुहरूका लागि धन्यवाद, तपाईंले विभाजनको वास्ता नगरी अनुप्रयोगमा जहाँसुकै सन्दर्भ डेटा पहुँच गर्न सक्नुहुन्छ।
• केडीई स्ट्रिम, केटेबल र ग्लोबलकेटेबल वस्तुहरू बीच जडान सम्भव छ।

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

→ पुस्तकको बारेमा थप विवरणहरू यहाँ पाउन सकिन्छ प्रकाशकको ​​वेबसाइट

→ Habrozhiteli को लागि कुपन प्रयोग गरेर 25% छुट - काफ्का स्ट्रिम्स

→ पुस्तकको कागजी संस्करणको लागि भुक्तानी गरेपछि, एक इलेक्ट्रोनिक पुस्तक इ-मेल मार्फत पठाइनेछ।

स्रोत: www.habr.com