សៀវភៅ "Kafka ស្ទ្រីមនៅក្នុងសកម្មភាព។ កម្មវិធី និងសេវាមីក្រូសម្រាប់ការងារក្នុងពេលវេលាជាក់ស្តែង"

សួស្តីអ្នកស្រុក Khabro! សៀវភៅនេះគឺសមរម្យសម្រាប់អ្នកអភិវឌ្ឍន៍ណាដែលចង់យល់អំពីដំណើរការខ្សែស្រឡាយ។ ការយល់ដឹងអំពីកម្មវិធីចែកចាយនឹងជួយអ្នកឱ្យយល់កាន់តែច្បាស់អំពី Kafka និង Kafka Stream ។ វាជាការល្អក្នុងការដឹងពីក្របខណ្ឌ Kafka ដោយខ្លួនឯង ប៉ុន្តែនេះមិនចាំបាច់ទេ៖ ខ្ញុំនឹងប្រាប់អ្នកនូវអ្វីគ្រប់យ៉ាងដែលអ្នកត្រូវការ។ អ្នកអភិវឌ្ឍន៍ Kafka ដែលមានបទពិសោធន៍ និងអ្នកថ្មីថ្មោងដូចគ្នានឹងរៀនពីរបៀបបង្កើតកម្មវិធីដំណើរការស្ទ្រីមគួរឱ្យចាប់អារម្មណ៍ដោយប្រើបណ្ណាល័យ Kafka Stream នៅក្នុងសៀវភៅនេះ។ អ្នកអភិវឌ្ឍន៍ Java កម្រិតមធ្យម និងកម្រិតខ្ពស់បានស្គាល់រួចហើយជាមួយនឹងគោលគំនិតដូចជា serialization នឹងរៀនអនុវត្តជំនាញរបស់ពួកគេដើម្បីបង្កើតកម្មវិធី Kafka Stream ។ កូដប្រភពរបស់សៀវភៅត្រូវបានសរសេរនៅក្នុង Java 8 ហើយប្រើប្រាស់វាក្យសម្ព័ន្ធកន្សោម Java 8 យ៉ាងសំខាន់ ដូច្នេះការដឹងពីរបៀបធ្វើការជាមួយមុខងារ lambda (សូម្បីតែនៅក្នុងភាសាសរសេរកម្មវិធីផ្សេងទៀត) នឹងមានប្រយោជន៍។

ដកស្រង់។ ៥.៣. ការប្រមូលផ្តុំនិងប្រតិបត្តិការបង្អួច

នៅក្នុងផ្នែកនេះ យើងនឹងបន្តស្វែងរកផ្នែកដែលជោគជ័យបំផុតនៃ Kafka Stream ។ រហូតមកដល់ពេលនេះយើងបានគ្របដណ្តប់ទិដ្ឋភាពដូចខាងក្រោមនៃ Kafka ស្ទ្រីម:

• ការបង្កើត topology ដំណើរការ;
• ការប្រើប្រាស់រដ្ឋនៅក្នុងកម្មវិធីស្ទ្រីម;
• អនុវត្តការតភ្ជាប់ស្ទ្រីមទិន្នន័យ;
• ភាពខុសគ្នារវាងស្ទ្រីមព្រឹត្តិការណ៍ (KStream) និងស្ទ្រីមអាប់ដេត (KTable)។

នៅក្នុងឧទាហរណ៍ខាងក្រោមយើងនឹងនាំយកធាតុទាំងអស់នេះមកជាមួយគ្នា។ អ្នក​ក៏​នឹង​រៀន​អំពី Windowing ដែល​ជា​មុខងារ​ដ៏​អស្ចារ្យ​មួយ​ទៀត​នៃ​កម្មវិធី​ស្ទ្រីម។ ឧទាហរណ៍ដំបូងរបស់យើងនឹងជាការបូកសរុបដ៏សាមញ្ញមួយ។

៥.៣.១. ការប្រមូលផ្តុំការលក់ភាគហ៊ុនតាមវិស័យឧស្សាហកម្ម

ការប្រមូលផ្តុំ និងការដាក់ជាក្រុមគឺជាឧបករណ៍សំខាន់នៅពេលធ្វើការជាមួយទិន្នន័យស្ទ្រីម។ ការ​ពិនិត្យ​លើ​កំណត់ត្រា​បុគ្គល​ដែល​ពួកគេ​ត្រូវ​បាន​ទទួល​ច្រើន​តែ​មិន​គ្រប់គ្រាន់។ ដើម្បីទាញយកព័ត៌មានបន្ថែមពីទិន្នន័យ ចាំបាច់ត្រូវដាក់ជាក្រុម និងបញ្ចូលគ្នា។

ក្នុងឧទាហរណ៍នេះ អ្នកនឹងដាក់សំលៀកបំពាក់របស់អ្នកជំនួញមួយថ្ងៃ ដែលត្រូវការតាមដានបរិមាណនៃការលក់ភាគហ៊ុនរបស់ក្រុមហ៊ុននៅក្នុងឧស្សាហកម្មមួយចំនួន។ ជាពិសេស អ្នកចាប់អារម្មណ៍លើក្រុមហ៊ុនទាំងប្រាំដែលមានការលក់ភាគហ៊ុនច្រើនជាងគេនៅក្នុងឧស្សាហកម្មនីមួយៗ។

ការប្រមូលផ្តុំបែបនេះនឹងតម្រូវឱ្យមានជំហានមួយចំនួនខាងក្រោមដើម្បីបកប្រែទិន្នន័យទៅជាទម្រង់ដែលចង់បាន (និយាយក្នុងន័យទូទៅ)។

1. បង្កើតប្រភពផ្អែកលើប្រធានបទដែលផ្សព្វផ្សាយព័ត៌មានជួញដូរភាគហ៊ុនឆៅ។ យើងនឹងត្រូវគូសផែនទីវត្ថុនៃប្រភេទ StockTransaction ទៅវត្ថុនៃប្រភេទ ShareVolume ។ ចំនុចនោះគឺថាវត្ថុ StockTransaction មានទិន្នន័យមេតានៃការលក់ ប៉ុន្តែយើងគ្រាន់តែត្រូវការទិន្នន័យអំពីចំនួនភាគហ៊ុនដែលត្រូវបានលក់ប៉ុណ្ណោះ។
2. ក្រុមទិន្នន័យ ShareVolume តាមនិមិត្តសញ្ញាភាគហ៊ុន។ នៅពេលដាក់ជាក្រុមតាមនិមិត្តសញ្ញា អ្នកអាចបង្រួមទិន្នន័យនេះទៅជាសរុបរងនៃបរិមាណលក់ភាគហ៊ុន។ វាគួរឱ្យកត់សម្គាល់ថាវិធីសាស្ត្រ KStream.groupBy ត្រឡប់ឧទាហរណ៍នៃប្រភេទ KGroupedStream ។ ហើយ​អ្នក​អាច​ទទួល​បាន​ឧទាហរណ៍ KTable ដោយ​ការ​ហៅ​ទៅ​កាន់​វិធីសាស្ត្រ KGroupedStream.reduce បន្ថែម។

តើអ្វីជាចំណុចប្រទាក់ KGroupedStream

វិធីសាស្ត្រ KStream.groupBy និង KStream.groupByKey ត្រឡប់ឧទាហរណ៍នៃ KGroupedStream ។ KGroupedStream គឺ​ជា​តំណាង​កម្រិត​មធ្យម​នៃ​ការ​ផ្សាយ​ព្រឹត្តិការណ៍​បន្ទាប់​ពី​ការ​ដាក់​ជា​ក្រុម​ដោយ​គ្រាប់ចុច។ វាមិនមានបំណងសម្រាប់ការងារដោយផ្ទាល់ជាមួយវាទាល់តែសោះ។ ជំនួសមកវិញ KGroupedStream ត្រូវបានប្រើសម្រាប់ប្រតិបត្តិការសរុប ដែលលទ្ធផលគឺតែងតែជា KTable ។ ហើយចាប់តាំងពីលទ្ធផលនៃប្រតិបត្តិការប្រមូលផ្តុំគឺជា KTable ហើយពួកគេប្រើប្រាស់ហាងរបស់រដ្ឋ វាអាចទៅរួចដែលថាមិនមែនរាល់ការអាប់ដេតជាលទ្ធផលត្រូវបានបញ្ជូនបន្តតាមបំពង់នោះទេ។

វិធីសាស្ត្រ KTable.groupBy ត្រឡប់ KGroupedTable ស្រដៀងគ្នា - តំណាងកម្រិតមធ្យមនៃស្ទ្រីមនៃបច្ចុប្បន្នភាព ដែលត្រូវបានដាក់ជាក្រុមឡើងវិញដោយគន្លឹះ។

ចូរយើងសម្រាកខ្លីមួយហើយមើលរូបភព។ 5.9 ដែលបង្ហាញពីអ្វីដែលយើងសម្រេចបាន។ topology នេះគួរតែស្គាល់អ្នករួចហើយ។

ឥឡូវនេះ សូមក្រឡេកមើលកូដសម្រាប់ topology នេះ (វាអាចត្រូវបានរកឃើញនៅក្នុងឯកសារ 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 សន្មត់ថាអ្នកបានផ្លាស់ប្តូរកូនសោដាក់ជាក្រុម ដូច្នេះទង់ការចែកចេញត្រូវបានកំណត់ទៅពិត។ ការ​អនុវត្ត​ការ​ចូលរួម ការ​រួម​បញ្ចូល​គ្នា​។
សេចក្តីសង្ខេប៖ នៅពេលណាដែលអាចធ្វើទៅបាន អ្នកគួរតែប្រើ GroupByKey ជាជាង GroupBy។

វាច្បាស់ណាស់នូវអ្វីដែល mapValues ​​និង groupBymethod ធ្វើ ដូច្នេះសូមក្រឡេកមើល sum() method (រកឃើញនៅក្នុង src/main/java/bbejeck/model/ShareVolume.java) (Listing 5.3)។

វិធីសាស្ត្រ ShareVolume.sum ត្រឡប់ចំនួនសរុបដែលកំពុងដំណើរការនៃបរិមាណលក់ភាគហ៊ុន ហើយលទ្ធផលនៃខ្សែសង្វាក់ទាំងមូលនៃការគណនាគឺជាវត្ថុ KTable . ឥឡូវនេះ អ្នកយល់ពីតួនាទីរបស់ KTable ហើយ។ នៅពេលដែលវត្ថុ ShareVolume មកដល់ វត្ថុ KTable ដែលត្រូវគ្នារក្សាទុកបច្ចុប្បន្នភាពបច្ចុប្បន្នចុងក្រោយបំផុត។ វាជារឿងសំខាន់ក្នុងការចងចាំថាការអាប់ដេតទាំងអស់ត្រូវបានឆ្លុះបញ្ចាំងនៅក្នុង shareVolumeKTable មុន ប៉ុន្តែមិនមែនទាំងអស់ត្រូវបានផ្ញើបន្ថែមទៀតនោះទេ។

បន្ទាប់មកយើងប្រើ KTable នេះដើម្បីប្រមូលផ្តុំ (តាមចំនួនភាគហ៊ុនដែលបានជួញដូរ) ដើម្បីទៅដល់ក្រុមហ៊ុនទាំងប្រាំដែលមានបរិមាណភាគហ៊ុនខ្ពស់បំផុតដែលត្រូវបានជួញដូរនៅក្នុងឧស្សាហកម្មនីមួយៗ។ សកម្មភាពរបស់យើងក្នុងករណីនេះនឹងស្រដៀងគ្នាទៅនឹងសកម្មភាពនៃការប្រមូលផ្តុំដំបូង។

1. អនុវត្តក្រុមមួយផ្សេងទៀតដោយប្រតិបត្តិការដើម្បីដាក់ក្រុមវត្ថុ ShareVolume នីមួយៗតាមឧស្សាហកម្ម។
2. ចាប់ផ្តើមសង្ខេបវត្ថុ ShareVolume ។ លើកនេះវត្ថុប្រមូលផ្តុំគឺជាជួរអាទិភាពទំហំថេរ។ នៅក្នុងជួរទំហំថេរនេះ មានតែក្រុមហ៊ុនទាំងប្រាំដែលមានចំនួនភាគហ៊ុនច្រើនជាងគេដែលត្រូវបានលក់ត្រូវបានរក្សាទុក។
3. គូសផែនទីជួរពីកថាខណ្ឌមុនទៅតម្លៃខ្សែអក្សរ ហើយត្រឡប់ភាគហ៊ុនកំពូលទាំងប្រាំដែលត្រូវបានជួញដូរច្រើនបំផុតតាមលេខតាមឧស្សាហកម្ម។
4. សរសេរលទ្ធផលជាទម្រង់ខ្សែអក្សរទៅប្រធានបទ។

នៅក្នុងរូបភព។ រូបភាព 5.10 បង្ហាញក្រាហ្វលំហូរទិន្នន័យ។ ដូចដែលអ្នកអាចឃើញជុំទីពីរនៃដំណើរការគឺសាមញ្ញណាស់។

ឥឡូវនេះ យើងមានការយល់ដឹងយ៉ាងច្បាស់អំពីរចនាសម្ព័ន្ធនៃដំណើរការជុំទីពីរនេះ យើងអាចងាកទៅរកកូដប្រភពរបស់វា (អ្នកនឹងឃើញវានៅក្នុងឯកសារ src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (ការចុះបញ្ជី 5.4) .

កម្មវិធីចាប់ផ្តើមនេះមានអថេរQueueថេរ។ នេះគឺជាវត្ថុផ្ទាល់ខ្លួនដែលជាអាដាប់ទ័រសម្រាប់ java.util.TreeSet ដែលត្រូវបានប្រើដើម្បីតាមដានលទ្ធផល N កំពូលតាមលំដាប់ចុះនៃភាគហ៊ុនដែលបានជួញដូរ។

អ្នកបានឃើញការហៅជាក្រុមដោយ និង mapValues ​​រួចហើយ ដូច្នេះយើងនឹងមិនចូលទៅក្នុងវាទេ (យើងកំពុងហៅវិធីសាស្ត្រ KTable.toStream ពីព្រោះវិធីសាស្ត្រ KTable.print ត្រូវបានបដិសេធ)។ ប៉ុន្តែអ្នកមិនទាន់បានឃើញកំណែ KTable នៃ aggregate() នៅឡើយទេ ដូច្នេះយើងនឹងចំណាយពេលបន្តិចដើម្បីពិភាក្សាអំពីវា។

ដូចដែលអ្នកចងចាំ អ្វីដែលធ្វើឱ្យ KTable មានភាពខុសប្លែកគ្នានោះគឺថា កំណត់ត្រាដែលមានគ្រាប់ចុចដូចគ្នាត្រូវបានគេចាត់ទុកថាជាការធ្វើបច្ចុប្បន្នភាព។ KTable ជំនួសធាតុចាស់ជាមួយនឹងធាតុថ្មី។ ការប្រមូលផ្តុំកើតឡើងតាមរបៀបស្រដៀងគ្នា៖ កំណត់ត្រាចុងក្រោយបំផុតដែលមានគន្លឹះដូចគ្នាត្រូវបានប្រមូលផ្តុំ។ នៅពេលដែលកំណត់ត្រាមកដល់ វាត្រូវបានបន្ថែមទៅថ្នាក់ FixedSizePriorityQueue ដោយប្រើ adder (ប៉ារ៉ាម៉ែត្រទីពីរក្នុងការហៅ method សរុប) ប៉ុន្តែប្រសិនបើ record ផ្សេងទៀតមានរួចហើយជាមួយនឹង key ដូចគ្នានោះ record ចាស់ត្រូវបានយកចេញដោយប្រើ subtractor (ប៉ារ៉ាម៉ែត្រទីបីនៅក្នុង ការហៅវិធីសាស្រ្តសរុប) ។

នេះមានន័យថាអ្នកប្រមូលផ្ដុំរបស់យើង FixedSizePriorityQueue មិនសរុបតម្លៃទាំងអស់ដោយប្រើសោមួយ ប៉ុន្តែរក្សាទុកផលបូកនៃបរិមាណនៃប្រភេទភាគហ៊ុនដែលជួញដូរភាគច្រើនបំផុត។ ធាតុចូលនីមួយៗមានចំនួនសរុបនៃភាគហ៊ុនដែលបានលក់រហូតមកដល់ពេលនេះ។ KTable នឹងផ្តល់ឱ្យអ្នកនូវព័ត៌មានអំពីភាគហ៊ុនរបស់ក្រុមហ៊ុនណាដែលបច្ចុប្បន្នត្រូវបានជួញដូរច្រើនជាងគេ ដោយមិនតម្រូវឱ្យមានការបង្រួបបង្រួមនៃការធ្វើបច្ចុប្បន្នភាពនីមួយៗឡើយ។

យើងបានរៀនធ្វើរឿងសំខាន់ពីរ៖

• តម្លៃ​ក្រុម​ក្នុង KTable ដោយ​កូនសោ​ទូទៅ;
• អនុវត្តប្រតិបត្តិការដែលមានប្រយោជន៍ដូចជាការប្រមូលផ្តុំ និងការប្រមូលផ្តុំលើតម្លៃដែលបានដាក់ជាក្រុមទាំងនេះ។

ការដឹងពីរបៀបអនុវត្តប្រតិបត្តិការទាំងនេះមានសារៈសំខាន់ក្នុងការយល់ដឹងពីអត្ថន័យនៃទិន្នន័យដែលផ្លាស់ទីតាមរយៈកម្មវិធី Kafka Stream និងការយល់ដឹងអំពីព័ត៌មានដែលវាផ្ទុក។

យើងក៏បាននាំយកនូវគោលគំនិតសំខាន់ៗមួយចំនួនដែលបានពិភាក្សាពីមុនមកក្នុងសៀវភៅនេះ។ នៅក្នុងជំពូកទី 4 យើងបានពិភាក្សាអំពីរបៀបដែលការអត់ឱនចំពោះកំហុស រដ្ឋក្នុងតំបន់មានសារៈសំខាន់សម្រាប់កម្មវិធីផ្សាយ។ ឧទាហរណ៍ទីមួយនៅក្នុងជំពូកនេះ បានបង្ហាញឱ្យឃើញពីមូលហេតុដែលរដ្ឋមូលដ្ឋានមានសារៈសំខាន់ណាស់ — វាអនុញ្ញាតឱ្យអ្នកតាមដានព័ត៌មានដែលអ្នកបានឃើញរួចហើយ។ ការចូលប្រើក្នុងតំបន់ជៀសវាងការពន្យារពេលបណ្តាញដែលធ្វើឱ្យកម្មវិធីកាន់តែដំណើរការ និងធន់នឹងកំហុស។

នៅពេលអនុវត្តប្រតិបត្តិការប្រមូល ឬប្រមូលផ្តុំណាមួយ អ្នកត្រូវតែបញ្ជាក់ឈ្មោះរបស់ហាងរដ្ឋ។ ប្រតិបតិ្តការរួម និងការប្រមូលផ្តុំត្រឡប់វត្ថុ KTable ហើយ KTable ប្រើកន្លែងផ្ទុករបស់រដ្ឋ ដើម្បីជំនួសលទ្ធផលចាស់ជាមួយថ្មី។ ដូចដែលអ្នកបានឃើញហើយ មិនមែនរាល់ការអាប់ដេតទាំងអស់ត្រូវបានបញ្ជូនតាមបំពង់ទេ ហើយនេះគឺសំខាន់ណាស់ ពីព្រោះប្រតិបត្តិការប្រមូលផ្តុំត្រូវបានរចនាឡើងដើម្បីបង្កើតព័ត៌មានសង្ខេប។ ប្រសិនបើអ្នកមិនអនុវត្តរដ្ឋមូលដ្ឋានទេ KTable នឹងបញ្ជូនលទ្ធផលសរុប និងលទ្ធផលសរុប។

បន្ទាប់ យើងនឹងពិនិត្យមើលប្រតិបត្តិការដូចជាការប្រមូលផ្តុំក្នុងរយៈពេលជាក់លាក់ណាមួយ - ដែលគេហៅថា ប្រតិបត្តិការបង្អួច។

៥.៣.២. ប្រតិបត្តិការបង្អួច

នៅក្នុងផ្នែកមុន យើងបានណែនាំអំពីការរំកិលរំកិល និងការបូកបញ្ចូលគ្នា។ កម្មវិធីនេះបានអនុវត្តការលក់ភាគហ៊ុនជាបន្តបន្ទាប់ដែលបន្តដោយការប្រមូលផ្តុំនៃភាគហ៊ុនចំនួនប្រាំដែលត្រូវបានជួញដូរច្រើនបំផុតនៅលើទីផ្សារប្តូរប្រាក់។

ជួនកាលការបូកសរុបបន្ត និងការបូកសរុបលទ្ធផលគឺចាំបាច់។ ហើយពេលខ្លះអ្នកត្រូវធ្វើប្រតិបត្តិការតែក្នុងរយៈពេលកំណត់ប៉ុណ្ណោះ។ ឧទាហរណ៍ គណនាចំនួនប្រតិបត្តិការប្តូរប្រាក់ត្រូវបានធ្វើឡើងជាមួយភាគហ៊ុនរបស់ក្រុមហ៊ុនជាក់លាក់មួយក្នុងរយៈពេល 10 នាទីចុងក្រោយ។ ឬតើអ្នកប្រើប្រាស់ប៉ុន្មាននាក់បានចុចលើផ្ទាំងផ្សាយពាណិជ្ជកម្មថ្មីក្នុងរយៈពេល 15 នាទីចុងក្រោយ។ កម្មវិធីមួយអាចអនុវត្តប្រតិបត្តិការបែបនេះច្រើនដង ប៉ុន្តែជាមួយនឹងលទ្ធផលដែលអនុវត្តចំពោះតែរយៈពេលដែលបានបញ្ជាក់ (បង្អួចពេលវេលា)។

រាប់ប្រតិបត្តិការប្តូរប្រាក់ដោយអ្នកទិញ

នៅក្នុងឧទាហរណ៍បន្ទាប់ យើងនឹងតាមដានប្រតិបត្តិការភាគហ៊ុននៅទូទាំងពាណិជ្ជករជាច្រើន - ទាំងអង្គការធំ ឬអ្នកហិរញ្ញវត្ថុបុគ្គលដ៏ឆ្លាតវៃ។

មានហេតុផលពីរដែលអាចកើតមានសម្រាប់ការតាមដាននេះ។ មួយក្នុងចំនោមពួកគេគឺត្រូវដឹងពីអ្វីដែលអ្នកដឹកនាំទីផ្សារកំពុងទិញ/លក់។ ប្រសិនបើអ្នកលេងធំៗ និងអ្នកវិនិយោគដ៏ស្មុគ្រស្មាញទាំងនេះមើលឃើញឱកាស វាសមហេតុផលក្នុងការធ្វើតាមយុទ្ធសាស្ត្ររបស់ពួកគេ។ មូលហេតុទីពីរគឺការចង់ឃើញសញ្ញាដែលអាចកើតមាននៃការជួញដូរខាងក្នុងខុសច្បាប់។ ដើម្បីធ្វើដូច្នេះបាន អ្នកនឹងត្រូវវិភាគការជាប់ទាក់ទងគ្នានៃការកើនឡើងនៃការលក់ដ៏ធំជាមួយនឹងសេចក្តីប្រកាសព័ត៌មានសំខាន់ៗ។

ការតាមដានបែបនេះមានជំហានដូចខាងក្រោមៈ

• បង្កើតស្ទ្រីមសម្រាប់ការអានពីប្រធានបទភាគហ៊ុន-ប្រតិបត្តិការ;
• ដាក់ក្រុមកំណត់ត្រាចូលដោយលេខសម្គាល់អ្នកទិញ និងនិមិត្តសញ្ញាភាគហ៊ុន។ ការ​ហៅ​ទៅ​ក្រុម​តាម​វិធីសាស្ត្រ​ត្រឡប់​ឧទាហរណ៍​នៃ​ថ្នាក់ KGroupedStream;
• វិធីសាស្ត្រ KGroupedStream.windowedBy ត្រឡប់ការស្ទ្រីមទិន្នន័យដែលកំណត់ទៅវិនដូពេលវេលា ដែលអនុញ្ញាតឱ្យមានការប្រមូលផ្តុំតាមបង្អួច។ អាស្រ័យលើប្រភេទបង្អួច ទាំង TimeWindowedKStream ឬ SessionWindowedKStream ត្រូវបានត្រឡប់។
• ចំនួនប្រតិបត្តិការសម្រាប់ប្រតិបត្តិការសរុប។ លំហូរទិន្នន័យតាមបង្អួចកំណត់ថាតើកំណត់ត្រាជាក់លាក់មួយត្រូវបានយកទៅក្នុងគណនីនៅក្នុងចំនួននេះដែរឬទេ។
• សរសេរលទ្ធផលទៅប្រធានបទ ឬបញ្ជូនវាទៅកុងសូលកំឡុងពេលអភិវឌ្ឍ។

topology នៃកម្មវិធីនេះគឺសាមញ្ញ ប៉ុន្តែរូបភាពច្បាស់លាស់របស់វានឹងមានប្រយោជន៍។ សូមក្រឡេកមើលរូបភព។ ៥.១១.

បន្ទាប់ យើងនឹងពិនិត្យមើលមុខងារនៃប្រតិបត្តិការបង្អួច និងកូដដែលត្រូវគ្នា។

ប្រភេទបង្អួច

មានបង្អួចបីប្រភេទនៅក្នុងស្ទ្រីម Kafka:

• សម័យប្រជុំ;
• "ដួល" (ដួល);
• រអិល/លោត។

តើមួយណាដែលត្រូវជ្រើសរើសអាស្រ័យលើតម្រូវការអាជីវកម្មរបស់អ្នក។ ផ្ទាំងវិនដូលោត និងលោតត្រូវបានកំណត់ពេលវេលា ខណៈពេលដែលវិនដូសម័យត្រូវបានកំណត់ដោយសកម្មភាពអ្នកប្រើប្រាស់ - រយៈពេលនៃវគ្គត្រូវបានកំណត់តែឯងដោយថាតើអ្នកប្រើប្រាស់សកម្មប៉ុណ្ណា។ រឿងចំបងដែលត្រូវចងចាំគឺថាប្រភេទបង្អួចទាំងអស់គឺផ្អែកលើត្រាកាលបរិច្ឆេទ/ពេលវេលានៃធាតុ មិនមែនម៉ោងប្រព័ន្ធទេ។

បន្ទាប់មក យើងអនុវត្ត topology របស់យើងជាមួយនឹងប្រភេទបង្អួចនីមួយៗ។ កូដពេញលេញនឹងត្រូវបានផ្តល់ឱ្យតែនៅក្នុងឧទាហរណ៍ដំបូងប៉ុណ្ណោះ សម្រាប់ប្រភេទផ្សេងទៀតនៃបង្អួច គ្មានអ្វីនឹងផ្លាស់ប្តូរទេលើកលែងតែប្រភេទនៃប្រតិបត្តិការបង្អួច។

បង្អួចសម័យ

បង្អួច​សម័យ​គឺ​ខុស​គ្នា​ខ្លាំង​ពី​ប្រភេទ​បង្អួច​ផ្សេង​ទៀត។ ពួកវាត្រូវបានកំណត់មិនច្រើនតាមពេលវេលាដូចជាសកម្មភាពរបស់អ្នកប្រើប្រាស់ (ឬសកម្មភាពរបស់អង្គភាពដែលអ្នកចង់តាមដាន)។ វិនដូសម័យត្រូវបានកំណត់ដោយរយៈពេលនៃភាពអសកម្ម។

រូបភាព 5.12 បង្ហាញពីគំនិតនៃបង្អួចសម័យ។ វគ្គតូចជាងនឹងបញ្ចូលគ្នាជាមួយវគ្គនៅខាងឆ្វេងរបស់វា។ ហើយវគ្គនៅខាងស្តាំនឹងដាច់ដោយឡែកពីគ្នា ព្រោះវាធ្វើតាមរយៈពេលដ៏យូរនៃភាពអសកម្ម។ វិនដូសម័យគឺផ្អែកលើសកម្មភាពរបស់អ្នកប្រើ ប៉ុន្តែប្រើត្រាកាលបរិច្ឆេទ/ពេលវេលាពីធាតុដើម្បីកំណត់ថាតើវគ្គណាដែលជាកម្មសិទ្ធិរបស់ធាតុ។

ការប្រើវិនដូសម័យដើម្បីតាមដានប្រតិបត្តិការភាគហ៊ុន

តោះប្រើវិនដូសម័យដើម្បីចាប់យកព័ត៌មានអំពីប្រតិបត្តិការប្តូរប្រាក់។ ការអនុវត្តនៃបង្អួចសម័យត្រូវបានបង្ហាញនៅក្នុងបញ្ជី 5.5 (ដែលអាចត្រូវបានរកឃើញនៅក្នុង src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java)។

អ្នកបានឃើញប្រតិបត្តិការភាគច្រើននៅក្នុង topology នេះរួចហើយ ដូច្នេះមិនចាំបាច់មើលពួកវាម្តងទៀតនៅទីនេះទេ។ ប៉ុន្តែក៏មានធាតុថ្មីៗជាច្រើននៅទីនេះផងដែរ ដែលឥឡូវនេះយើងនឹងពិភាក្សា។

ក្រុមណាមួយដោយប្រតិបត្តិការជាធម្មតាអនុវត្តប្រភេទនៃប្រតិបត្តិការប្រមូលផ្តុំមួយចំនួន (ការប្រមូលផ្តុំ ការបង្រួបបង្រួម ឬរាប់)។ អ្នក​អាច​អនុវត្ត​ការ​ប្រមូល​ផ្ដុំ​ដោយ​ចំនួន​សរុប​ដែល​កំពុង​ដំណើរការ ឬ​ការ​បូក​បញ្ចូល​បង្អួច ដែល​គិត​ដល់​កំណត់ត្រា​គណនី​ក្នុង​ចន្លោះ​ពេល​វេលា​ដែល​បាន​បញ្ជាក់។

លេខកូដនៅក្នុងបញ្ជី 5.5 រាប់ចំនួនប្រតិបត្តិការនៅក្នុងវិនដូវគ្គ។ នៅក្នុងរូបភព។ 5.13 សកម្មភាពទាំងនេះត្រូវបានវិភាគជាជំហានៗ។

ដោយការហៅទៅកាន់ windowedBy(SessionWindows.with(twentySeconds)) រហូតដល់(ដប់ប្រាំនាទី)) យើងបង្កើតវិនដូសម័យមួយដែលមានចន្លោះពេលអសកម្ម 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: ០០:១៥ .

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។

ពីអ្វីដែលបានពិពណ៌នានៅក្នុងផ្នែកនេះ វាគឺមានតំលៃចងចាំ nuances សំខាន់ៗដូចខាងក្រោម:

• វគ្គមិនមែនជាវិនដូទំហំថេរទេ។ រយៈពេលនៃវគ្គមួយត្រូវបានកំណត់ដោយសកម្មភាពក្នុងរយៈពេលដែលបានផ្តល់ឱ្យ;
• ការបោះត្រាកាលបរិច្ឆេទ/ពេលវេលានៅក្នុងទិន្នន័យកំណត់ថាតើព្រឹត្តិការណ៍នេះស្ថិតនៅក្នុងវគ្គដែលមានស្រាប់ ឬអំឡុងពេលទំនេរ។

បន្ទាប់យើងនឹងពិភាក្សាអំពីប្រភេទនៃបង្អួចបន្ទាប់ - បង្អួច "ដួលរលំ" ។

បង្អួច "ដួលរលំ"

បង្អួចដែលដួលរលំចាប់យកព្រឹត្តិការណ៍ដែលធ្លាក់ក្នុងរយៈពេលជាក់លាក់ណាមួយ។ ស្រមៃថាអ្នកត្រូវចាប់យកប្រតិបត្តិការភាគហ៊ុនទាំងអស់របស់ក្រុមហ៊ុនជាក់លាក់រៀងរាល់ 20 វិនាទី ដូច្នេះអ្នកប្រមូលព្រឹត្តិការណ៍ទាំងអស់ក្នុងអំឡុងពេលនោះ។ នៅចុងបញ្ចប់នៃចន្លោះពេល 20 វិនាទី បង្អួចវិលជុំវិញ ហើយផ្លាស់ទីទៅចន្លោះពេលសង្កេត 20 វិនាទីថ្មី។ រូបភាព 5.14 បង្ហាញពីស្ថានភាពនេះ។

ដូចដែលអ្នកអាចឃើញ ព្រឹត្តិការណ៍ទាំងអស់ដែលទទួលបានក្នុងរយៈពេល 20 វិនាទីចុងក្រោយត្រូវបានរួមបញ្ចូលនៅក្នុងបង្អួច។ នៅចុងបញ្ចប់នៃរយៈពេលនេះ បង្អួចថ្មីមួយត្រូវបានបង្កើត។

ការចុះបញ្ជី 5.6 បង្ហាញកូដដែលបង្ហាញពីការប្រើប្រាស់វិនដូដួលរលំដើម្បីចាប់យកប្រតិបត្តិការភាគហ៊ុនរៀងរាល់ 20 វិនាទី (រកឃើញក្នុង src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java)។

ជាមួយនឹងការផ្លាស់ប្តូរតូចមួយនេះទៅការហៅតាមវិធីសាស្ត្រ TimeWindows.of អ្នកអាចប្រើផ្ទាំងវិនដូលោតបាន។ ឧទាហរណ៍នេះមិនហៅវិធីសាស្ត្រ រហូតដល់() ទេ ដូច្នេះចន្លោះពេលរក្សាទុកលំនាំដើម 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 ។

យើងនឹងភ្ជាប់ប្រតិបត្តិការទាំងនេះជាមួយគ្នាដូចខាងក្រោម (កូដអាចរកបាននៅក្នុងឯកសារ 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 ត្រូវបានប្រើនៅក្នុងការកំណត់។ ផងដែរ ដោយប្រើការរាប់មុនដំបូង តារាងត្រូវបានបំពេញដោយកំណត់ត្រានៅដើមដំបូង។

ឥឡូវនេះយើងអាចបន្តទៅជំហានចុងក្រោយ - ការតភ្ជាប់។

ការភ្ជាប់ព័ត៌មានថ្មីៗជាមួយនឹងទិន្នន័យចំនួនប្រតិបត្តិការ

ការបង្កើតការតភ្ជាប់មិនពិបាកទេ។ យើងនឹងប្រើការចូលរួមខាងឆ្វេង ក្នុងករណីដែលមិនមានព័ត៌មានស្តុកទុកសម្រាប់ឧស្សាហកម្មដែលពាក់ព័ន្ធ (កូដចាំបាច់អាចរកបាននៅក្នុងឯកសារ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (ការចុះបញ្ជី 5.10)។

ប្រតិបត្តិករ leftJoin នេះគឺសាមញ្ញណាស់។ មិនដូចការភ្ជាប់នៅក្នុងជំពូកទី 4 ទេ វិធីសាស្ត្រ JoinWindow មិនត្រូវបានប្រើទេ ពីព្រោះនៅពេលអនុវត្តការភ្ជាប់ KStream-KTable មានធាតុចូលតែមួយនៅក្នុង KTable សម្រាប់សោនីមួយៗ។ ការតភ្ជាប់បែបនេះមិនត្រូវបានកំណត់ក្នុងពេលវេលាទេ៖ កំណត់ត្រាគឺនៅក្នុង KTable ឬអវត្តមាន។ ការសន្និដ្ឋានចំបង៖ ដោយប្រើវត្ថុ KTable អ្នកអាចបង្កើន KStream ជាមួយនឹងទិន្នន័យយោងដែលមិនសូវទាន់សម័យជាញឹកញាប់។

ឥឡូវនេះ យើងនឹងពិនិត្យមើលវិធីដែលមានប្រសិទ្ធភាពជាងមុន ដើម្បីបង្កើនព្រឹត្តិការណ៍ពី KStream។

៥.៣.៤. វត្ថុ GlobalKTable

ដូចដែលអ្នកអាចឃើញ មានតម្រូវការក្នុងការពង្រឹងការផ្សាយព្រឹត្តិការណ៍ ឬបន្ថែមបរិបទទៅពួកគេ។ នៅក្នុងជំពូកទី 4 អ្នកបានឃើញការតភ្ជាប់រវាងវត្ថុ KStream ពីរ ហើយនៅក្នុងផ្នែកមុន អ្នកបានឃើញការតភ្ជាប់រវាង KStream និង KTable ។ ក្នុង​ករណី​ទាំង​អស់​នេះ វា​ជា​ការ​ចាំបាច់​ក្នុង​ការ​បែងចែក​ចរន្ត​ទិន្នន័យ​ឡើង​វិញ​នៅ​ពេល​ផ្គូផ្គង​គ្រាប់ចុច​ទៅ​ប្រភេទ​ឬ​តម្លៃ​ថ្មី។ ពេលខ្លះការបែងចែកឡើងវិញត្រូវបានធ្វើឡើងយ៉ាងជាក់លាក់ ហើយពេលខ្លះ Kafka Stream ធ្វើវាដោយស្វ័យប្រវត្តិ។ ការបែងចែកឡើងវិញគឺចាំបាច់ ពីព្រោះគ្រាប់ចុចបានផ្លាស់ប្តូរ ហើយកំណត់ត្រាត្រូវតែបញ្ចប់នៅក្នុងផ្នែកថ្មី បើមិនដូច្នេះទេ ការតភ្ជាប់នឹងមិនអាចទៅរួចទេ (វាត្រូវបានពិភាក្សានៅក្នុងជំពូកទី 4 នៅក្នុងផ្នែក "ការបែងចែកទិន្នន័យឡើងវិញ" នៅក្នុងផ្នែករង 4.2.4) ។

ការបែងចែកឡើងវិញមានការចំណាយ

ការបែងចែកឡើងវិញតម្រូវឱ្យមានការចំណាយ - ការចំណាយធនធានបន្ថែមសម្រាប់ការបង្កើតប្រធានបទកម្រិតមធ្យម ការរក្សាទុកទិន្នន័យស្ទួននៅក្នុងប្រធានបទមួយផ្សេងទៀត; វាក៏មានន័យថាភាពយឺតយ៉ាវកើនឡើងដោយសារតែការសរសេរ និងការអានពីប្រធានបទនេះ។ បន្ថែមពីលើនេះ ប្រសិនបើអ្នកត្រូវការភ្ជាប់ទិដ្ឋភាព ឬវិមាត្រច្រើនជាងមួយ អ្នកត្រូវតែភ្ជាប់ការភ្ជាប់ ផែនទីកំណត់ត្រាជាមួយនឹងសោថ្មី ហើយដំណើរការដំណើរការបែងចែកឡើងវិញម្តងទៀត។

ការភ្ជាប់ទៅសំណុំទិន្នន័យតូចជាង

ក្នុងករណីខ្លះ បរិមាណនៃទិន្នន័យយោងដែលត្រូវភ្ជាប់គឺតូច ដូច្នេះច្បាប់ចម្លងពេញលេញរបស់វាអាចសមនឹងមូលដ្ឋានបានយ៉ាងងាយស្រួលនៅលើថ្នាំងនីមួយៗ។ សម្រាប់ស្ថានភាពដូចនេះ Kafka Stream ផ្តល់នូវថ្នាក់ GlobalKTable ។

GlobalKTable instances គឺមានតែមួយគត់ ពីព្រោះកម្មវិធីចម្លងទិន្នន័យទាំងអស់ទៅ nodes នីមួយៗ។ ហើយចាប់តាំងពីទិន្នន័យទាំងអស់មានវត្តមាននៅលើថ្នាំងនីមួយៗ វាមិនចាំបាច់បែងចែកការផ្សាយព្រឹត្តិការណ៍ដោយគន្លឹះទិន្នន័យយោងនោះទេ ដូច្នេះវាអាចប្រើបានសម្រាប់គ្រប់ភាគថាសទាំងអស់។ អ្នកក៏អាចបង្កើតការភ្ជាប់ដោយគ្មានសោដោយប្រើវត្ថុ GlobalKTable ។ តោះត្រឡប់ទៅឧទាហរណ៍មួយក្នុងចំនោមឧទាហរណ៍ពីមុន ដើម្បីបង្ហាញលក្ខណៈពិសេសនេះ។

ការភ្ជាប់វត្ថុ KStream ទៅវត្ថុ GlobalKTable

នៅក្នុងផ្នែករង 5.3.2 យើងបានអនុវត្តការប្រមូលផ្តុំបង្អួចនៃប្រតិបត្តិការប្តូរប្រាក់ដោយអ្នកទិញ។ លទ្ធផលនៃការប្រមូលផ្តុំនេះមើលទៅដូចនេះ៖

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

ខណៈពេលដែលលទ្ធផលទាំងនេះបានបម្រើគោលបំណង វានឹងមានប្រយោជន៍ជាងប្រសិនបើឈ្មោះរបស់អតិថិជន និងឈ្មោះក្រុមហ៊ុនពេញលេញត្រូវបានបង្ហាញផងដែរ។ ដើម្បីបន្ថែមឈ្មោះអតិថិជន និងឈ្មោះក្រុមហ៊ុន អ្នកអាចធ្វើការភ្ជាប់ធម្មតា ប៉ុន្តែអ្នកនឹងត្រូវធ្វើការគូសផែនទីគន្លឹះពីរ និងបែងចែកឡើងវិញ។ ជាមួយនឹង GlobalKTable អ្នកអាចជៀសវាងការចំណាយនៃប្រតិបត្តិការបែបនេះ។

ដើម្បីធ្វើដូច្នេះ យើងនឹងប្រើវត្ថុ countStream ពី Listing 5.11 (កូដដែលត្រូវគ្នាអាចរកបាននៅក្នុង src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) ហើយភ្ជាប់វាទៅវត្ថុ GlobalKTable ពីរ។

យើង​បាន​ពិភាក្សា​រឿង​នេះ​រួច​ហើយ ដូច្នេះ​ខ្ញុំ​នឹង​មិន​និយាយ​វា​ទៀត​ទេ។ ប៉ុន្តែខ្ញុំចំណាំថា កូដនៅក្នុងអនុគមន៍ toStream(.map) ត្រូវបានអរូបីទៅជាវត្ថុមុខងារ ជំនួសឱ្យកន្សោម lambda ក្នុងជួរសម្រាប់ជាប្រយោជន៍នៃការអាន។

ជំហានបន្ទាប់គឺដើម្បីប្រកាសឧទាហរណ៍ពីរនៃ GlobalKTable (កូដដែលបានបង្ហាញអាចត្រូវបានរកឃើញនៅក្នុងឯកសារ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (រាយបញ្ជី 5.12)។

សូមចំណាំថាឈ្មោះប្រធានបទត្រូវបានពិពណ៌នាដោយប្រើប្រភេទដែលបានរាប់បញ្ចូល។

ឥឡូវនេះយើងមានសមាសធាតុទាំងអស់រួចរាល់ហើយ អ្វីដែលនៅសេសសល់គឺត្រូវសរសេរកូដសម្រាប់ការតភ្ជាប់ (ដែលអាចត្រូវបានរកឃើញនៅក្នុងឯកសារ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (រាយបញ្ជី 5.13)។

ទោះបីជាមានការភ្ជាប់គ្នាពីរនៅក្នុងកូដនេះក៏ដោយ ក៏ពួកគេត្រូវបានជាប់ច្រវាក់ ដោយសារតែលទ្ធផលរបស់ពួកគេទាំងពីរមិនត្រូវបានប្រើប្រាស់ដោយឡែកពីគ្នា។ លទ្ធផលត្រូវបានបង្ហាញនៅចុងបញ្ចប់នៃប្រតិបត្តិការទាំងមូល។

នៅពេលអ្នកដំណើរការប្រតិបត្តិការ Join ខាងលើ អ្នកនឹងទទួលបានលទ្ធផលដូចនេះ៖

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

ខ្លឹមសារមិនបានផ្លាស់ប្តូរទេ ប៉ុន្តែលទ្ធផលទាំងនេះមើលទៅកាន់តែច្បាស់។

ប្រសិនបើអ្នករាប់ថយក្រោយទៅជំពូកទី 4 អ្នកបានឃើញរួចហើយនូវប្រភេទនៃការតភ្ជាប់ជាច្រើននៅក្នុងសកម្មភាព។ ពួកគេត្រូវបានរាយក្នុងតារាង។ ៥.២. តារាងនេះឆ្លុះបញ្ចាំងពីសមត្ថភាពតភ្ជាប់ជាកំណែ 5.2 នៃ Kafka Stream ។ អ្វីមួយអាចផ្លាស់ប្តូរនៅក្នុងការចេញផ្សាយនាពេលអនាគត។

ដើម្បី​បញ្ចប់​រឿង សូម​សង្ខេប​ពី​ចំណុច​មូលដ្ឋាន៖ អ្នក​អាច​ភ្ជាប់​ការ​ផ្សាយ​ព្រឹត្តិការណ៍ (KStream) និង​ធ្វើ​បច្ចុប្បន្នភាព​ការ​ផ្សាយ (KTable) ដោយ​ប្រើ​ស្ថានភាព​មូលដ្ឋាន។ ជាជម្រើស ប្រសិនបើទំហំនៃទិន្នន័យយោងមិនធំពេក អ្នកអាចប្រើវត្ថុ GlobalKTable ។ GlobalKTables ចម្លងភាគថាសទាំងអស់ទៅថ្នាំងកម្មវិធី Kafka Stream នីមួយៗ ដោយធានាថាទិន្នន័យទាំងអស់អាចរកបាន ដោយមិនគិតពីភាគថាសណាមួយដែលត្រូវគ្នានោះទេ។

បន្ទាប់ យើងនឹងឃើញលក្ខណៈពិសេស Kafka Stream អរគុណដែលយើងអាចសង្កេតមើលការផ្លាស់ប្តូររដ្ឋដោយមិនប្រើប្រាស់ទិន្នន័យពីប្រធានបទ Kafka។

៥.៣.៥. ស្ថានភាពដែលអាចសាកសួរបាន។

យើង​បាន​អនុវត្ត​ប្រតិបត្តិការ​មួយ​ចំនួន​រួច​ហើយ​ដែល​ពាក់ព័ន្ធ​នឹង​ស្ថានភាព​ ហើយ​តែង​តែ​បង្ហាញ​លទ្ធផល​ទៅ​កុងសូល (សម្រាប់គោលបំណង​អភិវឌ្ឍន៍) ឬ​សរសេរ​វា​ទៅ​ប្រធានបទ (សម្រាប់គោលបំណង​ផលិត)។ នៅពេលសរសេរលទ្ធផលទៅប្រធានបទ អ្នកត្រូវប្រើអតិថិជន Kafka ដើម្បីមើលពួកគេ។

ការអានទិន្នន័យពីប្រធានបទទាំងនេះអាចត្រូវបានចាត់ទុកថាជាប្រភេទនៃទិដ្ឋភាពជាក់ស្តែង។ សម្រាប់គោលបំណងរបស់យើង យើងអាចប្រើនិយមន័យនៃទិដ្ឋភាពជាក់ស្តែងពីវិគីភីឌា៖ “... វត្ថុមូលដ្ឋានទិន្នន័យរូបវន្តដែលមានលទ្ធផលនៃសំណួរ។ ឧទាហរណ៍ វាអាចជាច្បាប់ចម្លងមូលដ្ឋាននៃទិន្នន័យពីចម្ងាយ ឬសំណុំរងនៃជួរដេក និង/ឬជួរឈរនៃតារាង ឬលទ្ធផលចូលរួម ឬតារាងសង្ខេបដែលទទួលបានតាមរយៈការប្រមូលផ្តុំ" (https://en.wikipedia.org/wiki /materialized_view) ។

Kafka Stream ក៏អនុញ្ញាតឱ្យអ្នកដំណើរការសំណួរអន្តរកម្មនៅលើហាងរបស់រដ្ឋ ដែលអនុញ្ញាតឱ្យអ្នកអានដោយផ្ទាល់នូវទិដ្ឋភាពជាក់ស្តែងទាំងនេះ។ វាជាការសំខាន់ក្នុងការកត់សម្គាល់ថាសំណួរទៅកាន់ហាងរដ្ឋគឺជាប្រតិបត្តិការបានតែអានប៉ុណ្ណោះ។ នេះធានាថាអ្នកមិនចាំបាច់ព្រួយបារម្ភអំពីការធ្វើឱ្យស្ថានភាពមិនស្របគ្នាដោយចៃដន្យ ខណៈពេលដែលកម្មវិធីរបស់អ្នកកំពុងដំណើរការទិន្នន័យ។

សមត្ថភាពក្នុងការសាកសួរដោយផ្ទាល់នូវហាងរបស់រដ្ឋមានសារៈសំខាន់ណាស់។ នេះមានន័យថាអ្នកអាចបង្កើតកម្មវិធីផ្ទាំងគ្រប់គ្រងដោយមិនចាំបាច់ទាញយកទិន្នន័យពីអ្នកប្រើប្រាស់ Kafka ជាមុនសិន។ វាក៏បង្កើនប្រសិទ្ធភាពនៃកម្មវិធីផងដែរ ដោយសារតែមិនចាំបាច់សរសេរទិន្នន័យម្តងទៀត៖

• សូមអរគុណដល់មូលដ្ឋានទិន្នន័យ ពួកគេអាចចូលប្រើបានយ៉ាងឆាប់រហ័ស។
• ការចម្លងទិន្នន័យត្រូវបានលុបចោល ព្រោះវាមិនត្រូវបានសរសេរទៅកន្លែងផ្ទុកខាងក្រៅទេ។

រឿងសំខាន់ដែលខ្ញុំចង់ឱ្យអ្នកចងចាំគឺថាអ្នកអាចសួរស្ថានភាពដោយផ្ទាល់ពីក្នុងកម្មវិធីរបស់អ្នក។ ឱកាស​ដែល​ផ្តល់​ឱ្យ​អ្នក​មិន​អាច​ត្រូវ​បាន​គេ​និយាយ​លើស​ពី​នេះ​ទេ។ ជំនួសឱ្យការប្រើប្រាស់ទិន្នន័យពី Kafka និងរក្សាទុកកំណត់ត្រានៅក្នុងមូលដ្ឋានទិន្នន័យសម្រាប់កម្មវិធី អ្នកអាចសាកសួរហាងរដ្ឋជាមួយនឹងលទ្ធផលដូចគ្នា។ សំណួរផ្ទាល់ទៅកាន់ហាងរដ្ឋមានន័យថាកូដតិច (គ្មានអ្នកប្រើប្រាស់) និងកម្មវិធីតិច (មិនត្រូវការតារាងទិន្នន័យដើម្បីរក្សាទុកលទ្ធផល)។

យើងបានរៀបរាប់អំពីមូលដ្ឋានបន្តិចនៅក្នុងជំពូកនេះ ដូច្នេះយើងនឹងចាកចេញពីការពិភាក្សារបស់យើងអំពីសំណួរអន្តរកម្មប្រឆាំងនឹងហាងរបស់រដ្ឋសម្រាប់ពេលនេះ។ ប៉ុន្តែកុំបារម្ភ៖ នៅក្នុងជំពូកទី 9 យើងនឹងបង្កើតកម្មវិធីផ្ទាំងគ្រប់គ្រងដ៏សាមញ្ញមួយដែលមានសំណួរអន្តរកម្ម។ វានឹងប្រើឧទាហរណ៍មួយចំនួនពីជំពូកនេះ និងជំពូកមុន ដើម្បីបង្ហាញពីសំណួរអន្តរកម្ម និងរបៀបដែលអ្នកអាចបន្ថែមពួកវាទៅក្នុងកម្មវិធី Kafka Stream ។

សង្ខេប

• វត្ថុ KStream តំណាងឱ្យស្ទ្រីមនៃព្រឹត្តិការណ៍ ប្រៀបធៀបទៅនឹងការបញ្ចូលទៅក្នុងមូលដ្ឋានទិន្នន័យ។ វត្ថុ KTable តំណាងឱ្យស្ទ្រីមអាប់ដេត ច្រើនទៀតដូចជាការអាប់ដេតទៅកាន់មូលដ្ឋានទិន្នន័យ។ ទំហំរបស់វត្ថុ KTable មិនលូតលាស់ទេ កំណត់ត្រាចាស់ត្រូវបានជំនួសដោយថ្មី។
• វត្ថុ KTable ត្រូវបានទាមទារសម្រាប់ប្រតិបត្តិការប្រមូលផ្តុំ។
• ដោយប្រើប្រតិបត្តិការ windowing អ្នកអាចបំបែកទិន្នន័យសរុបទៅជាធុងពេលវេលា។
• សូមអរគុណដល់វត្ថុ GlobalKTable អ្នកអាចចូលប្រើទិន្នន័យយោងគ្រប់ទីកន្លែងនៅក្នុងកម្មវិធីដោយមិនគិតពីការបែងចែក។
• ការភ្ជាប់រវាងវត្ថុ KStream, KTable និង GlobalKTable គឺអាចធ្វើទៅបាន។

រហូតមកដល់ពេលនេះ យើងបានផ្តោតលើការកសាងកម្មវិធី Kafka Stream ដោយប្រើ KStream DSL កម្រិតខ្ពស់។ ទោះបីជាវិធីសាស្រ្តកម្រិតខ្ពស់អនុញ្ញាតឱ្យអ្នកបង្កើតកម្មវិធីយ៉ាងត្រឹមត្រូវ និងសង្ខេបក៏ដោយ ការប្រើប្រាស់វាតំណាងឱ្យការដោះដូរ។ ការធ្វើការជាមួយ DSL KStream មានន័យថាបង្កើនភាពសង្ខេបនៃកូដរបស់អ្នកដោយកាត់បន្ថយកម្រិតនៃការគ្រប់គ្រង។ នៅក្នុងជំពូកបន្ទាប់ យើងនឹងពិនិត្យមើល API node handler កម្រិតទាប ហើយសាកល្បងការដោះដូរផ្សេងទៀត។ កម្មវិធីនឹងមានរយៈពេលយូរជាងពីមុន ប៉ុន្តែយើងនឹងអាចបង្កើតថ្នាំងឧបករណ៍ដោះស្រាយស្ទើរតែទាំងអស់ដែលយើងប្រហែលជាត្រូវការ។

→ ព័ត៌មានលម្អិតបន្ថែមអំពីសៀវភៅអាចរកបាននៅ គេហទំព័ររបស់អ្នកបោះពុម្ពផ្សាយ

→ សម្រាប់ការបញ្ចុះតម្លៃ 25% Habrozhiteli ដោយប្រើប័ណ្ណ - ស្ទ្រីម Kafka

→ នៅពេលបង់ប្រាក់សម្រាប់កំណែក្រដាស សៀវភៅអេឡិចត្រូនិចនឹងត្រូវបានផ្ញើតាមអ៊ីមែល។

ប្រភព: www.habr.com