წიგნი „კაფკა ნაკადები მოქმედებაში. აპლიკაციები და მიკროსერვისები რეალურ დროში მუშაობისთვის"

გამარჯობა ხაბროს მოსახლეობავ! ეს წიგნი შესაფერისია ნებისმიერი დეველოპერისთვის, რომელსაც სურს გაიგოს ძაფების დამუშავება. განაწილებული პროგრამირების გაგება დაგეხმარებათ უკეთ გაიგოთ კაფკასა და კაფკას ნაკადები. კარგი იქნებოდა თავად კაფკას ჩარჩოს ცოდნა, მაგრამ ეს არ არის აუცილებელი: მე გეტყვით ყველაფერს, რაც გჭირდებათ. გამოცდილი კაფკას დეველოპერები და ახალბედები ისწავლიან როგორ შექმნან საინტერესო ნაკადის დამუშავების აპლიკაციები ამ წიგნის Kafka Streams ბიბლიოთეკის გამოყენებით. საშუალო და მოწინავე Java დეველოპერები, რომლებიც უკვე იცნობენ ცნებებს, როგორიცაა სერიალიზაცია, ისწავლიან გამოიყენონ თავიანთი უნარები Kafka Streams აპლიკაციების შესაქმნელად. წიგნის წყაროს კოდი დაწერილია Java 8-ში და მნიშვნელოვნად იყენებს Java 8 ლამბდა გამოხატვის სინტაქსს, ამიტომ ლამბდა ფუნქციებთან მუშაობის ცოდნა (თუნდაც სხვა პროგრამირების ენაზე) გამოდგება.

ამონაწერი. 5.3. აგრეგაციისა და ფანჯრების ოპერაციები

ამ განყოფილებაში ჩვენ გადავალთ კაფკას ნაკადების ყველაზე პერსპექტიული ნაწილების შესწავლაზე. აქამდე ჩვენ გავაშუქეთ კაფკას ნაკადების შემდეგი ასპექტები:

• დამუშავების ტოპოლოგიის შექმნა;
• მდგომარეობის გამოყენება სტრიმინგ აპლიკაციებში;
• მონაცემთა ნაკადის კავშირების შესრულება;
• განსხვავებები მოვლენის ნაკადებს (KStream) და განახლების ნაკადებს შორის (KTable).

შემდეგ მაგალითებში ჩვენ გავაერთიანებთ ყველა ამ ელემენტს. თქვენ ასევე შეიტყობთ windowing-ის, სტრიმინგის აპლიკაციების კიდევ ერთი შესანიშნავი თვისების შესახებ. ჩვენი პირველი მაგალითი იქნება მარტივი აგრეგაცია.

5.3.1. აქციების გაყიდვების აგრეგაცია მრეწველობის სექტორის მიხედვით

აგრეგაცია და დაჯგუფება სასიცოცხლო ინსტრუმენტებია სტრიმინგის მონაცემებთან მუშაობისას. ინდივიდუალური ჩანაწერების შესწავლა მათი მიღებისას ხშირად არასაკმარისია. მონაცემებიდან დამატებითი ინფორმაციის ამოსაღებად აუცილებელია მათი დაჯგუფება და გაერთიანება.

ამ მაგალითში თქვენ ჩაიცვამთ დღის ტრეიდერის კოსტუმს, რომელიც უნდა აკონტროლოს რამდენიმე ინდუსტრიის კომპანიების აქციების გაყიდვების მოცულობა. კონკრეტულად, თქვენ გაინტერესებთ ხუთი კომპანია, რომლებსაც აქვთ ყველაზე დიდი წილის გაყიდვები თითოეულ ინდუსტრიაში.

ასეთი აგრეგაცია დასჭირდება რამდენიმე ნაბიჯს, რათა მოხდეს მონაცემების სასურველ ფორმაში გადაყვანა (ზოგადად საუბრისას).

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, რომელიც გვიჩვენებს, რას მივაღწიეთ. ეს ტოპოლოგია უკვე ძალიან ნაცნობი უნდა იყოს თქვენთვის.

მოდით ახლა გადავხედოთ ამ ტოპოლოგიის კოდს (ის შეგიძლიათ იხილოთ ფაილში src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (ჩამონათვალი 5.2).

მოცემული კოდი გამოირჩევა ლაკონურობით და რამდენიმე სტრიქონში შესრულებული მოქმედებების დიდი მოცულობით. თქვენ შეიძლება შეამჩნიოთ რაღაც ახალი builder.stream მეთოდის პირველ პარამეტრში: მნიშვნელობა enum ტიპის AutoOffsetReset.EARLIEST (არსებობს ასევე LATEST), დაყენებული Consumed.withOffsetResetPolicy მეთოდის გამოყენებით. ჩამოთვლის ეს ტიპი შეიძლება გამოყენებულ იქნას ოფსეტური გადატვირთვის სტრატეგიის დასაზუსტებლად თითოეული KStream-ისთვის ან KTable-ისთვის და უპირატესობა ენიჭება ოფსეტური გადატვირთვის ვარიანტს კონფიგურაციიდან.

GroupByKey და GroupBy

KStream ინტერფეისს აქვს ჩანაწერების დაჯგუფების ორი მეთოდი: GroupByKey და GroupBy. ორივე აბრუნებს KGroupedTable-ს, ასე რომ თქვენ შეიძლება გაინტერესებთ რა განსხვავებაა მათ შორის და როდის გამოვიყენოთ რომელი?

GroupByKey მეთოდი გამოიყენება, როდესაც კლავიშები KStream-ში უკვე არ არის ცარიელი. და რაც მთავარია, დროშა „მოითხოვს ხელახლა დაყოფას“ არასოდეს დაყენებულა.

GroupBy მეთოდი ვარაუდობს, რომ თქვენ შეცვალეთ დაჯგუფების კლავიშები, ამიტომ გადანაწილების დროშა დაყენებულია true. შეერთების, აგრეგაციების და ა.შ. GroupBy მეთოდის შესრულება გამოიწვევს ავტომატურ ხელახლა დაყოფას.
შეჯამება: შეძლებისდაგვარად, თქვენ უნდა გამოიყენოთ GroupByKey და არა GroupBy.

გასაგებია, რას აკეთებენ mapValues ​​და groupBy მეთოდები, ასე რომ, მოდით გადავხედოთ sum() მეთოდს (ნაპოვნი src/main/java/bbejeck/model/ShareVolume.java) (განცხადება 5.3).

ShareVolume.sum მეთოდი აბრუნებს აქციების გაყიდვების მოცულობის მიმდინარე ჯამს და გამოთვლების მთელი ჯაჭვის შედეგი არის KTable ობიექტი. . ახლა თქვენ გესმით KTable-ის როლი. როდესაც ShareVolume ობიექტები ჩამოდის, შესაბამისი KTable ობიექტი ინახავს უახლეს მიმდინარე განახლებას. მნიშვნელოვანია გვახსოვდეს, რომ ყველა განახლება აისახება წინა shareVolumeKTable-ში, მაგრამ ყველა არ არის გაგზავნილი შემდგომში.

შემდეგი, ამ KTable-ის გამოყენებით, ჩვენ ვაგროვებთ (ვაჭრული აქციების რაოდენობის მიხედვით) მივიდეთ ხუთ კომპანიამდე, რომელთაც აქვთ აქციების ყველაზე მეტი მოცულობის ვაჭრობა თითოეულ ინდუსტრიაში. ჩვენი ქმედებები ამ შემთხვევაში მსგავსი იქნება პირველი აგრეგაციისთვის.

1. შეასრულეთ სხვა groupBy ოპერაცია, რომ დააჯგუფოთ ShareVolume ცალკეული ობიექტები ინდუსტრიის მიხედვით.
2. დაიწყეთ ShareVolume ობიექტების შეჯამება. ამჯერად აგრეგაციის ობიექტი არის ფიქსირებული ზომის პრიორიტეტული რიგი. ამ ფიქსირებული ზომის რიგში, მხოლოდ ხუთი კომპანია რჩება გაყიდული აქციების ყველაზე დიდი რაოდენობით.
3. წინა აბზაცის რიგები გადაიტანეთ სტრიქონის მნიშვნელობამდე და დააბრუნეთ ყველაზე გაყიდვადი აქციების ტოპ ხუთეული რიცხვების მიხედვით ინდუსტრიის მიხედვით.
4. ჩაწერეთ შედეგები სტრიქონში თემაში.

ნახ. ნახაზი 5.10 გვიჩვენებს მონაცემთა ნაკადის ტოპოლოგიის გრაფიკს. როგორც ხედავთ, დამუშავების მეორე რაუნდი საკმაოდ მარტივია.

ახლა, როდესაც ჩვენ გვაქვს მკაფიოდ გააზრებული დამუშავების მეორე რაუნდის სტრუქტურა, შეგვიძლია მივმართოთ მის საწყის კოდს (მას იპოვით ფაილში src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (ჩამონათვალი 5.4) .

ეს ინიციალატორი შეიცავს fixedQueue ცვლადს. ეს არის მორგებული ობიექტი, რომელიც არის ადაპტერი java.util.TreeSet-ისთვის, რომელიც გამოიყენება ზედა N შედეგების თვალყურის დევნებისთვის ვაჭრობის აქციების კლებადობით.

თქვენ უკვე ნახეთ groupBy და mapValues ​​ზარები, ამიტომ ჩვენ არ შევალთ მათზე (ჩვენ ვუწოდებთ KTable.toStream მეთოდს, რადგან KTable.print მეთოდი მოძველებულია). მაგრამ თქვენ ჯერ არ გინახავთ aggregate()-ის KTable ვერსია, ამიტომ ცოტა დროს დავხარჯავთ ამის განხილვაზე.

როგორც გახსოვთ, რაც განასხვავებს KTable-ს, არის ის, რომ ჩანაწერები იგივე კლავიშებით განიხილება განახლებად. KTable ცვლის ძველ ჩანაწერს ახლით. აგრეგაცია ხდება ანალოგიურად: ერთი და იგივე გასაღების მქონე უახლესი ჩანაწერები აგრეგირებულია. როდესაც ჩანაწერი ჩამოდის, მას ემატება FixedSizePriorityQueue კლასის ინსტანცია შემკრების გამოყენებით (მეორე პარამეტრი აგრეგატის მეთოდის გამოძახებაში), მაგრამ თუ სხვა ჩანაწერი უკვე არსებობს იმავე გასაღებით, მაშინ ძველი ჩანაწერი წაიშლება გამოკლების გამოყენებით (მესამე პარამეტრი აგრეგატული მეთოდის გამოძახება).

ეს ყველაფერი ნიშნავს, რომ ჩვენი აგრეგატორი, FixedSizePriorityQueue, არ აერთიანებს ყველა მნიშვნელობას ერთი გასაღებით, მაგრამ ინახავს N ყველაზე ვაჭრობის ტიპის აქციების რაოდენობის მოძრავ ჯამს. თითოეული შემომავალი ჩანაწერი შეიცავს აქამდე გაყიდული აქციების მთლიან რაოდენობას. KTable მოგაწვდით ინფორმაციას იმის შესახებ, თუ რომელი კომპანიების აქციებია ამჟამად ყველაზე მეტად ვაჭრობა, ყოველი განახლების მოძრავი აგრეგაციის საჭიროების გარეშე.

ჩვენ ვისწავლეთ ორი მნიშვნელოვანი საქმის გაკეთება:

• დააჯგუფეთ მნიშვნელობები KTable-ში საერთო გასაღებით;
• შეასრულეთ სასარგებლო ოპერაციები, როგორიცაა შეკრება და აგრეგაცია ამ დაჯგუფებულ მნიშვნელობებზე.

ამ ოპერაციების შესრულების ცოდნა მნიშვნელოვანია Kafka Streams-ის აპლიკაციაში მოძრავი მონაცემების მნიშვნელობის გასაგებად და იმის გასაგებად, თუ რა ინფორმაციას ატარებს იგი.

ჩვენ ასევე გავაერთიანეთ ამ წიგნში ადრე განხილული რამდენიმე ძირითადი კონცეფცია. მე-4 თავში განვიხილეთ, თუ რამდენად მნიშვნელოვანია დეფექტების შემწყნარებლობა, ლოკალური მდგომარეობა ნაკადის აპლიკაციისთვის. ამ თავში პირველმა მაგალითმა აჩვენა, თუ რატომ არის ადგილობრივი სახელმწიფო ასეთი მნიშვნელოვანი - ის საშუალებას გაძლევთ თვალყური ადევნოთ რა ინფორმაციას უკვე უნახავთ. ლოკალური წვდომა თავიდან აიცილებს ქსელის შეფერხებებს, რაც აპლიკაციას უფრო ეფექტურს და შეცდომებს გამძლეს ხდის.

ნებისმიერი შეკრების ან აგრეგაციის ოპერაციის შესრულებისას, თქვენ უნდა მიუთითოთ სახელმწიფო მაღაზიის სახელი. შეკრების და აგრეგაციის ოპერაციები აბრუნებს KTable ინსტანციას და KTable იყენებს მდგომარეობის შენახვას ძველი შედეგების ახლით ჩანაცვლებისთვის. როგორც ხედავთ, ყველა განახლება არ იგზავნება მილსადენში და ეს მნიშვნელოვანია, რადგან აგრეგაციის ოპერაციები შექმნილია შემაჯამებელი ინფორმაციის წარმოებისთვის. თუ არ გამოიყენებთ ადგილობრივ მდგომარეობას, KTable გადააგზავნის ყველა აგრეგაციისა და შეკრების შედეგებს.

შემდეგი, ჩვენ განვიხილავთ ოპერაციების შესრულებას, როგორიცაა აგრეგაცია გარკვეული დროის განმავლობაში - ე.წ. windowing ოპერაციები.

5.3.2. ფანჯრის ოპერაციები

წინა განყოფილებაში ჩვენ წარმოვადგინეთ მოცურების კონვოლუცია და აგრეგაცია. აპლიკაციამ შეასრულა აქციების გაყიდვების მოცულობის უწყვეტი დაგროვება, რასაც მოჰყვა ბირჟაზე ყველაზე მეტად ვაჭრობის ხუთი აქციის აგრეგაცია.

ზოგჯერ საჭიროა ასეთი უწყვეტი აგრეგაცია და შედეგების შერწყმა. და ზოგჯერ თქვენ გჭირდებათ ოპერაციების შესრულება მხოლოდ გარკვეული პერიოდის განმავლობაში. მაგალითად, გამოთვალეთ რამდენი გაცვლითი ტრანზაქცია განხორციელდა კონკრეტული კომპანიის აქციებით ბოლო 10 წუთის განმავლობაში. ან რამდენმა მომხმარებელმა დააწკაპუნა ახალ სარეკლამო ბანერზე ბოლო 15 წუთის განმავლობაში. აპლიკაციას შეუძლია რამდენჯერმე შეასრულოს ასეთი ოპერაციები, მაგრამ შედეგებით, რომლებიც ვრცელდება მხოლოდ დროის განსაზღვრულ პერიოდებზე (დროის ფანჯრები).

მყიდველის მიერ გაცვლითი ოპერაციების დათვლა

შემდეგ მაგალითში ჩვენ თვალყურს ვადევნებთ საფონდო ტრანზაქციებს რამდენიმე ტრეიდერში — მსხვილ ორგანიზაციებს ან ჭკვიან ინდივიდუალურ ფინანსისტებს.

ამ თვალთვალის ორი შესაძლო მიზეზი არსებობს. ერთ-ერთი მათგანია იმის ცოდნა, თუ რას ყიდულობენ/ყიდიან ბაზრის ლიდერები. თუ ეს დიდი მოთამაშეები და დახვეწილი ინვესტორები ხედავენ შესაძლებლობას, აზრი აქვს დაიცვას მათი სტრატეგია. მეორე მიზეზი არის არალეგალური ინსაიდერული ვაჭრობის შესაძლო ნიშნების გამოვლენის სურვილი. ამისათვის თქვენ უნდა გაანალიზოთ გაყიდვების დიდი მწვერვალების კორელაცია მნიშვნელოვან პრესრელიზებთან.

ასეთი თვალთვალი შედგება შემდეგი ნაბიჯებისგან:

• საფონდო ტრანზაქციების თემიდან წასაკითხად ნაკადის შექმნა;
• შემომავალი ჩანაწერების დაჯგუფება მყიდველის ID და აქციების სიმბოლოს მიხედვით. groupBy მეთოდის გამოძახებით აბრუნებს KGroupedStream კლასის ინსტანციას;
• KGroupedStream.windowedBy მეთოდი აბრუნებს მონაცემთა ნაკადს, რომელიც შემოიფარგლება დროის ფანჯრით, რომელიც იძლევა ფანჯრების აგრეგაციის საშუალებას. ფანჯრის ტიპის მიხედვით, ბრუნდება ან TimeWindowedKStream ან SessionWindowedKStream;
• ტრანზაქციის რაოდენობა აგრეგაციის ოპერაციისთვის. ფანჯრიანი მონაცემთა ნაკადი განსაზღვრავს, იქნება თუ არა გათვალისწინებული კონკრეტული ჩანაწერი ამ დათვლაში;
• შედეგების ჩაწერა თემაზე ან მათი კონსოლში გამოტანა განვითარების დროს.

ამ აპლიკაციის ტოპოლოგია მარტივია, მაგრამ მისი ნათელი სურათი სასარგებლო იქნება. მოდით შევხედოთ ნახ. 5.11.

შემდეგი, ჩვენ გადავხედავთ ფანჯრის ოპერაციების ფუნქციონირებას და შესაბამის კოდს.

ფანჯრების ტიპები

Kafka Streams-ში სამი ტიპის ფანჯარაა:

• სესიური;
• "ჩავარდნა";
• სრიალი/ხტუნვა.

რომელი აირჩიოს, დამოკიდებულია თქვენი ბიზნესის მოთხოვნებზე. ცურვისა და გადახტომის ფანჯრები დროში შეზღუდულია, ხოლო სესიის ფანჯრები შეზღუდულია მომხმარებლის აქტივობით - სესიის(ების) ხანგრძლივობა განისაზღვრება მხოლოდ იმით, თუ რამდენად აქტიურია მომხმარებელი. მთავარია გვახსოვდეს, რომ ფანჯრის ყველა ტიპი ეფუძნება ჩანაწერების თარიღს/დროს და არა სისტემის დროს.

შემდეგი, ჩვენ ვახორციელებთ ჩვენს ტოპოლოგიას ფანჯრების თითოეულ ტიპთან. სრული კოდი მოცემულია მხოლოდ პირველ მაგალითში; სხვა ტიპის ფანჯრებისთვის არაფერი შეიცვლება, გარდა ფანჯრის ოპერაციის ტიპისა.

სესიის ფანჯრები

სესიის ფანჯრები ძალიან განსხვავდება ყველა სხვა ტიპის ფანჯრებისგან. ისინი შეზღუდულია არა იმდენად დროით, რამდენადაც მომხმარებლის აქტივობით (ან იმ სუბიექტის აქტივობით, რომლის თვალყურის დევნებაც გსურთ). სესიის ფანჯრები შემოიფარგლება უმოქმედობის პერიოდებით.

სურათი 5.12 ასახავს სესიის ფანჯრების კონცეფციას. მცირე სესია გაერთიანდება სესიასთან მის მარცხნივ. და სხდომა მარჯვნივ იქნება ცალკე, რადგან ის მოჰყვება უმოქმედობის ხანგრძლივ პერიოდს. სესიის ფანჯრები დაფუძნებულია მომხმარებლის აქტივობაზე, მაგრამ გამოიყენეთ ჩანაწერებიდან თარიღის/დროის მარკები, რათა დაადგინოთ რომელ სესიას ეკუთვნის ჩანაწერი.

სესიის ფანჯრების გამოყენება საფონდო ტრანზაქციების თვალყურის დევნებისთვის

მოდით გამოვიყენოთ სესიის ფანჯრები გაცვლითი ტრანზაქციების შესახებ ინფორმაციის მისაღებად. სესიის ფანჯრების განხორციელება ნაჩვენებია სიაში 5.5 (რომელიც შეგიძლიათ იხილოთ src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java-ში).

თქვენ უკვე ნახეთ ოპერაციების უმეტესობა ამ ტოპოლოგიაში, ამიტომ არ არის საჭირო მათი ხელახლა ყურება აქ. მაგრამ აქ ასევე არის რამდენიმე ახალი ელემენტი, რომელზეც ახლა განვიხილავთ.

ნებისმიერი groupBy ოპერაცია, როგორც წესი, ასრულებს რაიმე სახის აგრეგაციის ოპერაციას (აგრეგაცია, შეკრება ან დათვლა). თქვენ შეგიძლიათ შეასრულოთ ან კუმულაციური აგრეგაცია გაშვებული ჯამით, ან ფანჯრის აგრეგაცია, რომელიც ითვალისწინებს ჩანაწერებს განსაზღვრულ დროის ფანჯარაში.

5.5 ჩამონათვალის კოდი ითვლის ტრანზაქციების რაოდენობას სესიის ფანჯრებში. ნახ. 5.13 ეს ქმედებები გაანალიზებულია ეტაპობრივად.

windowedBy(SessionWindows.with(twentySeconds).until(fifteenMiutes)) გამოძახებით ჩვენ ვქმნით სესიის ფანჯარას უმოქმედობის ინტერვალით 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.

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 მეთოდის გამოძახებით, შეგიძლიათ გამოიყენოთ ცვალებად ფანჯარა. ეს მაგალითი არ იძახებს სანამ() მეთოდს, ამიტომ გამოყენებული იქნება ნაგულისხმევი შეკავების ინტერვალი 24 საათის განმავლობაში.

დაბოლოს, დროა გადავიდეთ ფანჯრის ბოლო ვარიანტზე - "hopping" windows-ზე.

მოცურების ("ხტომა") ფანჯრები

მოცურების/სრიალო ფანჯრები წაშლილი ფანჯრების მსგავსია, მაგრამ მცირე განსხვავებით. მოცურების ფანჯრები არ დაელოდებიან დროის ინტერვალის დასრულებას, სანამ შექმნით ახალ ფანჯარას ბოლო მოვლენების დასამუშავებლად. ისინი იწყებენ ახალ გამოთვლებს ფანჯრის ხანგრძლივობაზე ნაკლები ლოდინის ინტერვალის შემდეგ.

საილუსტრაციოდ განსხვავებების საილუსტრაციოდ ჩამოვარდნილ და მხტუნავ ფანჯრებს შორის, დავუბრუნდეთ საფონდო ბირჟაზე ოპერაციების დათვლის მაგალითს. ჩვენი მიზანი ჯერ კიდევ არის ტრანზაქციების რაოდენობის დათვლა, მაგრამ არ გვინდა დაველოდოთ მთელი დროის განმავლობაში მრიცხველის განახლებამდე. ამის ნაცვლად, ჩვენ ვაახლებთ მრიცხველს უფრო მოკლე ინტერვალებით. მაგალითად, ჩვენ მაინც დავთვლით ტრანზაქციის რაოდენობას ყოველ 20 წამში, მაგრამ ვაახლებთ მრიცხველს ყოველ 5 წამში, როგორც ნაჩვენებია ნახ. 5.15. ამ შემთხვევაში, ჩვენ მივიღებთ სამი შედეგის ფანჯარას გადახურვის მონაცემებით.

ჩამონათვალი 5.7 აჩვენებს მოცურების ფანჯრების განსაზღვრის კოდს (ნაპოვნია src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

თაღლითური ფანჯარა შეიძლება გარდაიქმნას გადახტომის ფანჯარაში, ზარის დამატებით advanceBy() მეთოდზე. ნაჩვენები მაგალითში, შენახვის ინტერვალი არის 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 ობიექტი 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) (Listing 5.9).

აღსანიშნავია, რომ არ არის საჭირო Serde ობიექტების მითითება, რადგან პარამეტრებში გამოიყენება სტრიქონები Serdes. ასევე, ადრეული აღრიცხვის გამოყენებით, ცხრილი თავიდანვე ივსება ჩანაწერებით.

ახლა ჩვენ შეგვიძლია გადავიდეთ საბოლოო ეტაპზე - დაკავშირება.

ახალი ამბების განახლებების დაკავშირება ტრანზაქციის დათვლის მონაცემებთან

კავშირის შექმნა არ არის რთული. ჩვენ გამოვიყენებთ მარცხენა შეერთებას იმ შემთხვევაში, თუ არ არის საფონდო სიახლეები შესაბამისი ინდუსტრიისთვის (აუცილებელი კოდი შეგიძლიათ იხილოთ ფაილში src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.10).

მარცხენაJoin ოპერატორი საკმაოდ მარტივია. მე-4 თავში შეერთებისგან განსხვავებით, JoinWindow მეთოდი არ გამოიყენება, რადგან KStream-KTable შეერთების შესრულებისას, KTable-ში მხოლოდ ერთი ჩანაწერია თითოეული გასაღებისთვის. ასეთი კავშირი დროში შეზღუდული არ არის: ჩანაწერი არის KTable-ში ან არ არსებობს. მთავარი დასკვნა: KTable ობიექტების გამოყენებით შეგიძლიათ KStream-ის გამდიდრება ნაკლებად ხშირად განახლებული საცნობარო მონაცემებით.

ახლა ჩვენ გადავხედავთ KStream-ის ღონისძიებების გამდიდრების უფრო ეფექტურ გზას.

5.3.4. GlobalKTable ობიექტები

როგორც ხედავთ, საჭიროა ღონისძიების ნაკადების გამდიდრება ან მათ კონტექსტის დამატება. მე-4 თავში თქვენ იხილეთ კავშირები KStream-ის ორ ობიექტს შორის, ხოლო წინა ნაწილში თქვენ ნახეთ კავშირი KStream-სა და KTable-ს შორის. ყველა ამ შემთხვევაში, აუცილებელია მონაცემთა ნაკადის ხელახლა დაყოფა კლავიშების ახალ ტიპზე ან მნიშვნელობაზე შედგენისას. ზოგჯერ გადანაწილება ხდება ცალსახად, ზოგჯერ კი Kafka Streams ამას ავტომატურად აკეთებს. ხელახალი დაყოფა აუცილებელია, რადგან კლავიშები შეიცვალა და ჩანაწერები უნდა დასრულდეს ახალ განყოფილებებში, წინააღმდეგ შემთხვევაში კავშირი შეუძლებელი იქნება (ეს განხილული იყო მე-4 თავში, ქვეთავში „მონაცემების ხელახალი დაყოფა“ განყოფილებაში 4.2.4).

ხელახლა დაყოფას აქვს ფასი

ხელახალი დაყოფა მოითხოვს ხარჯებს - დამატებითი რესურსის ხარჯები შუალედური თემების შესაქმნელად, დუბლიკატი მონაცემების სხვა თემაში შესანახად; ეს ასევე ნიშნავს გაზრდილ შეყოვნებას ამ თემიდან წერისა და კითხვის გამო. გარდა ამისა, თუ გჭირდებათ ერთზე მეტი ასპექტის ან განზომილების შეერთება, თქვენ უნდა დააკავშიროთ შეერთებები, ჩანაწერების რუკა ახალი კლავიშებით და ხელახლა განახორციელოთ დაყოფის პროცესი.

დაკავშირება მცირე მონაცემთა ნაკრებებთან

ზოგიერთ შემთხვევაში, დასაკავშირებელი საცნობარო მონაცემების მოცულობა შედარებით მცირეა, ამიტომ მისი სრული ასლები ადვილად შეიძლება მოერგოს ადგილობრივ თითოეულ კვანძს. მსგავსი სიტუაციებისთვის, Kafka Streams უზრუნველყოფს GlobalKTable კლასს.

GlobalKTable ინსტანციები უნიკალურია, რადგან აპლიკაცია იმეორებს ყველა მონაცემს თითოეულ კვანძზე. და რადგან ყველა მონაცემი იმყოფება თითოეულ კვანძზე, არ არის საჭირო მოვლენის ნაკადის დაყოფა საცნობარო მონაცემთა გასაღებით ისე, რომ ის ხელმისაწვდომი იყოს ყველა დანაყოფისთვის. თქვენ ასევე შეგიძლიათ გააკეთოთ უწყვეტი შეერთებები GlobalKTable ობიექტების გამოყენებით. მოდით დავუბრუნდეთ ერთ-ერთ წინა მაგალითს ამ მახასიათებლის საჩვენებლად.

KStream ობიექტების დაკავშირება GlobalKTable ობიექტებთან

5.3.2 ქვეთავში ჩვენ შევასრულეთ მყიდველების მიერ გაცვლითი ოპერაციების ფანჯრის აგრეგაცია. ამ აგრეგაციის შედეგები ასე გამოიყურებოდა:

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

მიუხედავად იმისა, რომ ეს შედეგები ემსახურებოდა მიზანს, უფრო სასარგებლო იქნებოდა მომხმარებლის სახელი და კომპანიის სრული სახელიც რომ ყოფილიყო ნაჩვენები. მომხმარებლის სახელისა და კომპანიის სახელის დასამატებლად, შეგიძლიათ გააკეთოთ ნორმალური შეერთება, მაგრამ თქვენ დაგჭირდებათ ორი გასაღების რუკების გაკეთება და ხელახალი დაყოფა. GlobalKTable-ით შეგიძლიათ თავიდან აიცილოთ ასეთი ოპერაციების ღირებულება.

ამისათვის ჩვენ გამოვიყენებთ countStream ობიექტს ჩამონათვალიდან 5.11 (შესაბამისი კოდი შეგიძლიათ იხილოთ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) და დავუკავშიროთ მას GlobalKTable ორ ობიექტთან.

ჩვენ უკვე ვისაუბრეთ ამაზე, ამიტომ აღარ გავიმეორებ. მაგრამ მე აღვნიშნავ, რომ toStream().map ფუნქციის კოდი აბსტრაქტულია ფუნქციის ობიექტში, ნაცვლად inline ლამბდა გამოხატვისა, წაკითხვის მიზნით.

შემდეგი ნაბიჯი არის GlobalKTable-ის ორი ეგზემპლარის გამოცხადება (ნაჩვენები კოდი შეგიძლიათ იხილოთ ფაილში src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (ჩამონათვალი 5.12).

გთხოვთ გაითვალისწინოთ, რომ თემების სახელები აღწერილია ჩამოთვლილი ტიპების გამოყენებით.

ახლა, როცა ყველა კომპონენტი მზად გვაქვს, რჩება მხოლოდ დაკავშირების კოდის ჩაწერა (რომელიც შეგიძლიათ იხილოთ ფაილში src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Listing 5.13).

მიუხედავად იმისა, რომ ამ კოდში ორი შეერთებაა, ისინი მიჯაჭვულია, რადგან არცერთი შედეგი არ გამოიყენება ცალკე. შედეგები ნაჩვენებია მთელი ოპერაციის ბოლოს.

როდესაც აწარმოებთ ზემოთ შეერთების ოპერაციას, მიიღებთ შემდეგ შედეგებს:

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

არსი არ შეცვლილა, მაგრამ ეს შედეგები უფრო ნათლად გამოიყურება.

თუ უკუსვლით ითვლით მე-4 თავს, თქვენ უკვე ნახეთ რამდენიმე ტიპის კავშირი მოქმედებაში. ისინი ჩამოთვლილია ცხრილში. 5.2. ეს ცხრილი ასახავს Kafka Streams-ის 1.0.0 ვერსიის დაკავშირების შესაძლებლობებს; შეიძლება რაღაც შეიცვალოს მომავალ გამოშვებებში.

საქმის დასასრულებლად, მოდით შევაჯამოთ საფუძვლები: შეგიძლიათ დააკავშიროთ მოვლენის ნაკადები (KStream) და განაახლოთ ნაკადები (KTable) ადგილობრივი მდგომარეობის გამოყენებით. გარდა ამისა, თუ საცნობარო მონაცემების ზომა არ არის ძალიან დიდი, შეგიძლიათ გამოიყენოთ GlobalKTable ობიექტი. GlobalKTables იმეორებს ყველა დანაყოფის თითოეულ Kafka Streams-ის აპლიკაციის კვანძს, რაც უზრუნველყოფს ყველა მონაცემის ხელმისაწვდომობას, მიუხედავად იმისა, თუ რომელ დანაყოფს შეესაბამება გასაღები.

შემდეგ ჩვენ ვიხილავთ Kafka Streams ფუნქციას, რომლის წყალობითაც შეგვიძლია დავაკვირდეთ მდგომარეობის ცვლილებებს კაფკას თემის მონაცემების მოხმარების გარეშე.

5.3.5. საეჭვო მდგომარეობა

ჩვენ უკვე შევასრულეთ რამდენიმე ოპერაცია, რომელიც მოიცავს მდგომარეობას და ყოველთვის ვატანთ შედეგებს კონსოლში (განვითარების მიზნებისთვის) ან ვწერთ მათ თემაში (წარმოების მიზნებისთვის). როდესაც წერთ შედეგებს თემაზე, თქვენ უნდა გამოიყენოთ კაფკას მომხმარებელი მათი სანახავად.

ამ თემებიდან მონაცემების წაკითხვა შეიძლება ჩაითვალოს მატერიალიზებულ შეხედულებებად. ჩვენი მიზნებისთვის, ჩვენ შეგვიძლია ვიკიპედიიდან გამოვიყენოთ მატერიალიზებული ხედის განმარტება: „... ფიზიკური მონაცემთა ბაზის ობიექტი, რომელიც შეიცავს მოთხოვნის შედეგებს. მაგალითად, ეს შეიძლება იყოს დისტანციური მონაცემების ლოკალური ასლი, ან ცხრილის რიგების და/ან სვეტების ქვეჯგუფი ან შეერთების შედეგები, ან შემაჯამებელი ცხრილი, რომელიც მიღებულია აგრეგაციის გზით“ (https://en.wikipedia.org/wiki /მატერიალიზებული_ხედვა).

Kafka Streams ასევე საშუალებას გაძლევთ განახორციელოთ ინტერაქტიული მოთხოვნები სახელმწიფო მაღაზიებში, რაც საშუალებას გაძლევთ პირდაპირ წაიკითხოთ ეს მატერიალიზებული ხედები. მნიშვნელოვანია აღინიშნოს, რომ მოთხოვნა სახელმწიფო მაღაზიაში არის მხოლოდ წაკითხვის ოპერაცია. ეს უზრუნველყოფს იმას, რომ თქვენ არ უნდა ინერვიულოთ, რომ შემთხვევითი მდგომარეობა არათანმიმდევრულია, სანამ თქვენი აპლიკაცია ამუშავებს მონაცემებს.

მნიშვნელოვანია სახელმწიფო მაღაზიების უშუალო შეკითხვის შესაძლებლობა. ეს ნიშნავს, რომ თქვენ შეგიძლიათ შექმნათ საინფორმაციო დაფის აპლიკაციები კაფკას მომხმარებლისგან მონაცემების მოპოვების გარეშე. ეს ასევე ზრდის აპლიკაციის ეფექტურობას, იმის გამო, რომ არ არის საჭირო მონაცემების ხელახლა ჩაწერა:

• მონაცემთა ლოკალიზაციის წყალობით, მათ შეუძლიათ სწრაფად წვდომა;
• მონაცემთა დუბლირება აღმოფხვრილია, რადგან ის არ იწერება გარე საცავში.

მთავარი, რაც მინდა გახსოვდეთ, არის ის, რომ თქვენ შეგიძლიათ პირდაპირ მოითხოვოთ მდგომარეობა თქვენი აპლიკაციის შიგნიდან. შესაძლებლობები, რაც ეს გაძლევთ, არ შეიძლება გადაჭარბებული იყოს. იმის ნაცვლად, რომ მოიხმაროთ კაფკას მონაცემები და შეინახოთ ჩანაწერები აპლიკაციის მონაცემთა ბაზაში, შეგიძლიათ იგივე შედეგით მოითხოვოთ სახელმწიფო მაღაზიები. პირდაპირი მოთხოვნები სახელმწიფო მაღაზიებში ნიშნავს ნაკლებ კოდს (მომხმარებლის გარეშე) და ნაკლებ პროგრამულ უზრუნველყოფას (არ არის საჭირო მონაცემთა ბაზის ცხრილი შედეგების შესანახად).

ჩვენ ამ თავში საკმაოდ ბევრი ადგილი განვიხილეთ, ამიტომ სახელმწიფო მაღაზიების მიმართ ინტერაქტიული მოთხოვნების განხილვას ამ დროისთვის დავტოვებთ. მაგრამ არ ინერვიულოთ: მე-9 თავში ჩვენ შევქმნით მარტივ დაფის აპლიკაციას ინტერაქტიული მოთხოვნებით. ის გამოიყენებს ზოგიერთ მაგალითს ამ და წინა თავებიდან ინტერაქტიული მოთხოვნების დემონსტრირებისთვის და როგორ შეგიძლიათ დაამატოთ ისინი Kafka Streams აპლიკაციებში.

რეზიუმე

• KStream ობიექტები წარმოადგენს მოვლენების ნაკადებს, რომლებიც შედარებულია მონაცემთა ბაზაში ჩანართებთან. KTable ობიექტები წარმოადგენს განახლების ნაკადებს, უფრო ჰგავს მონაცემთა ბაზის განახლებებს. KTable ობიექტის ზომა არ იზრდება, ძველი ჩანაწერები იცვლება ახლით.
• KTable ობიექტები საჭიროა აგრეგაციის ოპერაციებისთვის.
• windowing ოპერაციების გამოყენებით, შეგიძლიათ დაყოთ აგრეგირებული მონაცემები დროის თაიგულებად.
• GlobalKTable ობიექტების წყალობით, თქვენ შეგიძლიათ წვდომა საცნობარო მონაცემებზე აპლიკაციის ნებისმიერ ადგილას, დანაწევრების მიუხედავად.
• შესაძლებელია კავშირები KStream, KTable და GlobalKTable ობიექტებს შორის.

აქამდე ჩვენ ორიენტირებული ვიყავით Kafka Streams აპლიკაციების შექმნაზე მაღალი დონის KStream DSL-ის გამოყენებით. მიუხედავად იმისა, რომ მაღალი დონის მიდგომა საშუალებას გაძლევთ შექმნათ სუფთა და ლაკონური პროგრამები, მისი გამოყენება წარმოადგენს კომპრომისს. DSL KStream-თან მუშაობა ნიშნავს თქვენი კოდის ლაკონურობის გაზრდას კონტროლის ხარისხის შემცირებით. შემდეგ თავში ჩვენ გადავხედავთ დაბალი დონის დამმუშავებლის კვანძის API-ს და შევეცდებით სხვა კომპრომისებს. პროგრამები უფრო გრძელი იქნება, ვიდრე ადრე იყო, მაგრამ ჩვენ შევძლებთ შევქმნათ თითქმის ნებისმიერი დამმუშავებელი კვანძი, რომელიც შეიძლება დაგვჭირდეს.

→ წიგნის შესახებ მეტი ინფორმაცია შეგიძლიათ იხილოთ აქ გამომცემლის ვებსაიტზე

→ ჰაბროჟითელზე 25% ფასდაკლება კუპონით - კაფკას ნაკადები

→ წიგნის ქაღალდის ვერსიის გადახდის შემდეგ ელექტრონული წიგნი გაიგზავნება ელექტრონული ფოსტით.

წყარო: www.habr.com