کتاب "کافکا په عمل کې جریان لري. د ریښتیني وخت کار لپاره غوښتنلیکونه او مایکرو خدمات"

سلامونه دخبرو اوسیدونکو! دا کتاب د هر هغه پراختیا کونکي لپاره مناسب دی چې غواړي د تار پروسس پوه شي. د ویشل شوي برنامه پوهیدل به تاسو سره د کافکا او کافکا جریانونو په ښه پوهیدو کې مرسته وکړي. دا به ښه وي چې پخپله د کافکا چوکاټ وپیژنئ ، مګر دا اړینه نده: زه به تاسو ته هرڅه ووایم چې تاسو ورته اړتیا لرئ. تجربه لرونکي کافکا پراختیا کونکي او نوي نوي کسان به په دې کتاب کې د کافکا سټیمز کتابتون په کارولو سره د زړه پورې جریان پروسس کولو غوښتنلیکونو رامینځته کولو څرنګوالي زده کړي. متوسط ​​​​او پرمختللي جاوا پراختیا کونکي دمخه د سیریل کولو په څیر مفاهیمو سره آشنا دي د کافکا سټریمونو غوښتنلیکونو رامینځته کولو لپاره به د دوی مهارتونو پلي کول زده کړي. د کتاب سرچینې کوډ په جاوا 8 کې لیکل شوی او د جاوا 8 لامبډا بیان ترکیب کې د پام وړ کار اخلي ، نو پدې پوهیدل چې څنګه د لامبډا فنکشن سره کار وکړي (حتی په بل پروګرامینګ ژبه کې) به په کار وي.

اقتباس. 5.3. د راټولولو او کړکۍ عملیات

په دې برخه کې، موږ به د کافکا سټریمونو خورا امید لرونکي برخې وپلټئ. تر دې دمه موږ د کافکا جریان لاندې اړخونه پوښلي دي:

• د پروسس کولو ټوپولوژي رامینځته کول؛
• د سټیمینګ غوښتنلیکونو کې د دولت کارول؛
• د ډیټا جریان اړیکې ترسره کول؛
• د پیښې جریانونو (KStream) او تازه جریانونو (KTable) ترمنځ توپیر.

په لاندې مثالونو کې به موږ دا ټول عناصر یوځای راوړو. تاسو به د وینډوز په اړه هم زده کړئ، د سټیمینګ غوښتنلیکونو بله غوره ځانګړتیا. زموږ لومړی مثال به یو ساده مجموعه وي.

5.3.1. د صنعت سکتور لخوا د سټاک پلور راټولول

راټولول او ګروپ کول حیاتي وسیلې دي کله چې د سټیمینګ ډیټا سره کار کوي. د انفرادي ریکارډونو ازموینه لکه څنګه چې دوی ترلاسه شوي اکثرا کافي ندي. د معلوماتو څخه د اضافي معلوماتو استخراج لپاره، دا اړینه ده چې ډله ایز او یوځای کړئ.

په دې مثال کې، تاسو به د یوې ورځې سوداګر جامې واچوئ څوک چې اړتیا لري په څو صنعتونو کې د شرکتونو سټاک پلور حجم تعقیب کړي. په ځانګړې توګه، تاسو د پنځو شرکتونو سره علاقه لرئ چې په هر صنعت کې د ونډو ترټولو لوی پلور لري.

دا ډول راټولول به لاندې څو مرحلو ته اړتیا ولري ترڅو ډاټا په مطلوب شکل کې وژباړي (په عمومي شرایطو کې خبرې کول).

1. د موضوع پر بنسټ سرچینه جوړه کړئ چې د خام سټاک سوداګرۍ معلومات خپروي. موږ باید د StockTransaction ډوله څیز د ShareVolume ډول اعتراض سره نقشه کړو. ټکی دا دی چې د سټاک لیږد اعتراض د پلور میټاډاټا لري، مګر موږ یوازې د پلورل شوي ونډو شمیرې په اړه معلوماتو ته اړتیا لرو.
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. دواړه د KGrouped Table بیرته راګرځوي، نو تاسو شاید حیران یاست چې د دوی ترمنځ توپیر څه دی او کله چې کوم یو وکاروئ؟

د GroupByKey طریقه کارول کیږي کله چې په KStream کې کیلي لا دمخه خالي نه وي. او تر ټولو مهم، د "بیا ویشلو ته اړتیا لري" بیرغ هیڅکله نه و ټاکل شوی.

د ګروپ بای میتود داسې انګیرل کیږي چې تاسو د ګروپ کولو کلیدونه بدل کړي، نو د بیا ویش بیرغ ریښتیا ته ټاکل شوی. د ګروپ بای میتود څخه وروسته یوځای کیدل ، راټولول او نور ترسره کول به د اتوماتیک بیا ویش پایله وي.
لنډیز: هرکله چې امکان ولري، تاسو باید د GroupBy پرځای GroupByKey وکاروئ.

دا روښانه ده چې د نقشې ارزښتونه او ګروپ بای میتودونه څه کوي، نو راځئ چې د sum() میتود (په src/main/java/bbejeck/model/ShareVolume.java کې موندل شوي) ته یو نظر وکړو (5.3 لیست کول).

د ShareVolume.sum طریقه د سټاک پلور حجم روان مجموعه بیرته راګرځوي، او د ټول محاسبې سلسله پایله د KTable څیز دی . اوس تاسو پوهیږئ چې د KTable رول لوبوي. کله چې د ShareVolume څیزونه راشي، د KTable اړونده توکي وروستی اوسنی اوسمهال ذخیره کوي. دا مهمه ده چې په یاد ولرئ چې ټول تازه معلومات په مخکیني شریکولومکیټبل کې منعکس شوي، مګر ټول نور نه لیږل کیږي.

بیا، د دې KTable په کارولو سره، موږ ټول (د سوداګرۍ د ونډو د شمیر له مخې) د پنځو شرکتونو ته ورسیږو چې په هر صنعت کې د ونډو ترټولو لوړ حجم لري. په دې قضیه کې زموږ کړنې به د لومړي راټولولو لپاره ورته وي.

1. د صنعت لخوا د انفرادي شریک حجم توکو ګروپ کولو لپاره د بل ګروپ لخوا عملیات ترسره کړئ.
2. د ShareVolume څیزونو لنډیز پیل کړئ. دا ځل د راټولولو اعتراض د ثابت اندازې لومړیتوب کتار دی. په دې ټاکل شوي اندازې کتار کې، یوازې پنځه شرکتونه ساتل کیږي چې د ونډو ترټولو لوی مقدار پلورل شوي.
3. کتارونه د مخکینۍ پراګراف څخه د تار ارزښت ته نقشه کړئ او د صنعت لخوا د شمیرې له مخې غوره پنځه خورا سوداګر شوي سټاک بیرته راستانه کړئ.
4. پایلې د موضوع لپاره د تار په بڼه ولیکئ.

په انځور کې. 5.10 شکل د ډیټا جریان ټوپولوژي ګراف ښیې. لکه څنګه چې تاسو لیدلی شئ، د پروسس دوهم پړاو خورا ساده دی.

اوس چې موږ د پروسس د دې دویم پړاو جوړښت په اړه روښانه پوهه لرو، موږ کولی شو د هغې سرچینې کوډ ته وګرځو (تاسو به یې په فایل کې ومومئ src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (5.4 لیست کول) .

دا پیل کونکی یو ثابت قطار متغیر لري. دا یو دودیز څیز دی چې د java.util.TreeSet لپاره اډاپټر دی چې د پلور شوي ونډو په نزولي ترتیب کې د پورتنۍ N پایلو تعقیب لپاره کارول کیږي.

تاسو لا دمخه د ګروپ بی او میپ ویلیو کالونه لیدلي دي، نو موږ به هغه ته لاړ نه شو (موږ د KTable.toStream میتود ته زنګ وهو ځکه چې د KTable.print میتود رد شوی دی). مګر تاسو تراوسه د Aggregate() KTable نسخه نه ده لیدلې، نو موږ به لږ وخت په دې بحث وکړو.

لکه څنګه چې تاسو په یاد ولرئ، هغه څه چې KTable توپیر کوي دا دی چې د ورته کیلي سره ریکارډونه تازه ګڼل کیږي. KTable پخوانۍ ننوتنه د نوي سره بدلوي. مجموعه په ورته ډول پیښیږي: د ورته کیلي سره وروستي ریکارډونه راټول شوي. کله چې یو ریکارډ راشي، دا د اډډر په کارولو سره د FixedSizePriorityQueue ټولګي مثال کې اضافه کیږي (په ټولیز میتود کال کې دوهم پیرامیټر)، مګر که بل ریکارډ دمخه د ورته کیلي سره شتون ولري، نو زوړ ریکارډ د فرعي کولو په کارولو سره لرې کیږي (په دریم پیرامیټر کې. د ټولیز میتود کال).

دا ټول پدې معنی دي چې زموږ جمع کونکی، FixedSizePriorityQueue، ټول ارزښتونه د یوې کیلي سره نه راټولوي، مګر د N ډیری سوداګریز ډولونو مقدارونو اندازه ذخیره کوي. هر راتلونکی داخله تر اوسه پورې د پلور شویو ونډو ټولټال شمیر لري. KTable به تاسو ته د دې په اړه معلومات درکړي چې د کومو شرکتونو ونډې اوس مهال تر ټولو ډیر تجارت کیږي، پرته له دې چې د هر تازه کولو رولینګ راټولولو ته اړتیا ولري.

موږ دوه مهم کارونه زده کړل:

• په KTable کې د ګروپ ارزښتونه د یو عام کیلي په واسطه؛
• ګټور عملیات ترسره کړئ لکه په دې ګروپ شوي ارزښتونو کې رول اپ او راټولول.

د دې عملیاتو د ترسره کولو څرنګوالي پوهیدل مهم دي چې د کافکا سټیمز غوښتنلیک له لارې د ډیټا حرکت کولو معنی پوهیدل او پدې پوهیدل چې کوم معلومات لري.

موږ په دې کتاب کې مخکې له مخکې بحث شوي ځینې کلیدي مفکورې هم راټولې کړې دي. په څلورم څپرکي کې، موږ بحث وکړ چې څنګه د غلطۍ زغم، محلي دولت د سټیمینګ غوښتنلیک لپاره مهم دی. په دې څپرکي کې لومړۍ بېلګه وښودله چې ولې محلي دولت دومره مهم دی - دا تاسو ته وړتیا درکوي چې هغه معلومات تعقیب کړئ چې تاسو دمخه لیدلي دي. ځایی لاسرسی د شبکې ځنډ څخه مخنیوی کوي، غوښتنلیک ډیر فعال او د خطا په وړاندې مقاومت کوي.

کله چې کوم رول اپ یا د راټولولو عملیات ترسره کوئ، تاسو باید د دولتي پلورنځي نوم مشخص کړئ. د رول اپ او راټولولو عملیات د KTable مثال بیرته راګرځوي، او KTable د پخوانیو پایلو سره د نوي سره بدلولو لپاره دولتي ذخیره کاروي. لکه څنګه چې تاسو لیدلي، ټول تازه معلومات د پایپ لاین لاندې نه لیږل کیږي، او دا مهم دی ځکه چې د راټولولو عملیات د لنډیز معلوماتو تولید لپاره ډیزاین شوي. که تاسو د محلي حالت غوښتنه نه کوئ، KTable به ټول راټولول او رول اپ پایلې وړاندې کړي.

بیا، موږ به د عملیاتو ترسره کولو ته وګورو لکه د یوې ټاکلې مودې دننه راټولول - د کړکۍ عملیات په نوم یادیږي.

5.3.2. کړکۍ عملیات

په تیره برخه کې، موږ د سلیډینګ کنولوشن او جمع کول معرفي کړل. غوښتنلیک د سټاک پلور دوامداره رول اپ ترسره کړ چې وروسته په تبادله کې د پنځو خورا سوداګریزو سټاکونو راټولول.

ځینې ​​​​وختونه دا ډول دوامداره راټولول او د پایلو راټولول اړین دي. او ځینې وختونه تاسو اړتیا لرئ یوازې د یوې ټاکلې مودې لپاره عملیات ترسره کړئ. د مثال په توګه، محاسبه کړئ چې په تیرو 10 دقیقو کې د یو ځانګړي شرکت ونډو سره څومره تبادله شوي. یا په تیرو 15 دقیقو کې څومره کاروونکو د نوي اعلاناتو بینر باندې کلیک کړی. یو غوښتنلیک ممکن دا ډول عملیات څو ځله ترسره کړي، مګر د پایلو سره چې یوازې د ټاکل شوي وخت (وخت وینډوز) لپاره پلي کیږي.

د پیرودونکي لخوا د تبادلې لیږد شمیرل

په راتلونکي مثال کې، موږ به د ډیری سوداګرو په اوږدو کې د سټاک لیږدونه تعقیب کړو - یا لوی سازمانونه یا سمارټ انفرادي تمویل کونکي.

د دې تعقیب لپاره دوه احتمالي دلیلونه شتون لري. یو له دوی څخه دا دی چې پوه شي چې د بازار مشران څه پیرود / پلوري. که دا لوی لوبغاړي او پیچلي پانګه اچوونکي فرصت وګوري، نو دا د دوی ستراتیژی تعقیبولو لپاره معنی لري. دوهم دلیل د غیرقانوني داخلي سوداګرۍ احتمالي نښې موندلو هیله ده. د دې کولو لپاره، تاسو اړتیا لرئ چې د مهمو مطبوعاتي اعلامیو سره د لوی پلور سپیکونو ارتباط تحلیل کړئ.

دا ډول تعقیب لاندې مرحلې لري:

• د سټاک لیږد موضوع څخه د لوستلو لپاره یو جریان رامینځته کول؛
• د پیرودونکي ID او سټاک سمبول لخوا د راتلونکو ریکارډونو ګروپ کول. د ګروپ بی طریقه ته زنګ وهل د KGrouped Stream ټولګي یوه بیلګه راګرځوي؛
• د KGroupedStream.windowedBy طریقه د وخت کړکۍ پورې محدود ډیټا جریان بیرته راګرځوي، کوم چې د کړکۍ راټولولو ته اجازه ورکوي. د کړکۍ ډول پورې اړه لري، یا د وخت وینډوز سټریم یا د سیشن وینډوډ سټریم بیرته راستنیږي؛
• د راټولولو عملیاتو لپاره د راکړې ورکړې شمیره. د کړکۍ شوي ډیټا جریان ټاکي چې ایا یو ځانګړی ریکارډ پدې شمیره کې په پام کې نیول شوی؛
• یوې موضوع ته پایلې لیکل یا د پراختیا په جریان کې کنسول ته راوړل.

د دې غوښتنلیک ټوپولوژي ساده ده، مګر د دې روښانه انځور به ګټور وي. راځئ چې انځور ته یو نظر وکړو. 5.11.

بیا، موږ به د کړکۍ عملیاتو فعالیت او اړونده کوډ وګورو.

د کړکۍ ډولونه

د کافکا په جریان کې درې ډوله کړکۍ شتون لري:

• غونډه
• "ټمبېدل"؛
• سلیډنګ/لپ کول.

کوم یو غوره کول ستاسو د سوداګرۍ اړتیاو پورې اړه لري. د ټمبلینګ او جمپ کولو کړکۍ وخت محدود دی، پداسې حال کې چې د سیشن کړکۍ د کارونکي فعالیت لخوا محدودې دي — د ناستې موده یوازې د کارونکي څومره فعاله لخوا ټاکل کیږي. د یادولو لپاره اصلي شی دا دی چې د کړکۍ ټول ډولونه د ننوتلو نیټې / وخت ټامپونو پراساس دي ، نه د سیسټم وخت.

بیا، موږ د هرې کړکۍ ډولونو سره زموږ ټوپولوژي پلي کوو. بشپړ کوډ به یوازې په لومړي مثال کې ورکړل شي؛ د نورو ډولونو وینډوز لپاره به د کړکۍ عملیاتو ډول پرته هیڅ شی بدل نشي.

د ناستې کړکۍ

د سیشن کړکۍ د نورو ټولو ډولونو کړکیو څخه خورا توپیر لري. دوی د وخت لخوا دومره محدود ندي لکه د کارونکي فعالیت (یا د هغه ادارې فعالیت چې تاسو یې تعقیب کول غواړئ). د سیشن کړکۍ د غیرفعالیت دورې لخوا محدود شوي.

5.12 شکل د سیشن کړکۍ مفهوم روښانه کوي. کوچنۍ ناسته به د هغې کیڼ اړخ ته د ناستې سره یوځای شي. او په ښي خوا کې ناسته به جلا وي ځکه چې دا د اوږدې مودې غیر فعالیت تعقیبوي. د سیشن کړکۍ د کارونکي فعالیت پراساس دي، مګر د ننوتلو نیټې/وخت ټاپهونه کاروي ترڅو معلومه کړي چې ننوتنه د کومې ناستې پورې اړه لري.

د سټاک لیږدونو تعقیب لپاره د سیشن وینډوز کارول

راځئ چې د سیشن کړکۍ وکاروو ترڅو د تبادلې لیږد په اړه معلومات ترلاسه کړو. د سیشن وینډوز پلي کول په لیست 5.5 کې ښودل شوي (کوم چې په src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java کې موندل کیدی شي).

تاسو لا دمخه په دې ټوپولوژي کې ډیری عملیات لیدلي، نو دلته د دوی بیاکتنې ته اړتیا نشته. مګر دلته یو شمیر نوي عناصر هم شتون لري، چې موږ به یې اوس بحث وکړو.

د هرې ډلې لخوا عملیات عموما یو ډول د راټولولو عملیات ترسره کوي (مجموعه، رول اپ، یا شمیرل). تاسو کولی شئ یا د چلولو مجموعې سره مجموعي مجموعه ترسره کړئ، یا د کړکۍ راټولول، کوم چې د ټاکل شوي وخت کړکۍ کې حساب ریکارډونه په پام کې نیسي.

په لست 5.5 کې کوډ د سیشن کړکۍ کې د معاملو شمیره شمیري. په انځور کې. 5.13 دا کړنې ګام په ګام تحلیل کیږي.

د windowedBy(SessionWindows.with(wentySeconds)) تر (پنځلس دقیقو) پورې زنګ وهلو سره موږ د ناستې کړکۍ د 20 ثانیو د غیر فعالیت وقفې او د 15 دقیقو دوام وقفې سره رامینځته کوو. د 20 ثانیو بې کاره وقفه پدې معنی ده چې په غوښتنلیک کې به هر هغه ننوتل شامل وي چې د اوسنۍ ناستې پای یا پیل په 20 ثانیو کې اوسنۍ (فعال) ناستې ته راځي.

بیا، موږ مشخص کوو چې د ناستې په کړکۍ کې د راټولولو عملیات باید ترسره شي - پدې حالت کې، شمیرل. که چیرې راتلونکی داخله د غیر فعالیت کړکۍ څخه بهر راشي (د نیټې / وخت ټاپه دواړو خواوو) ، غوښتنلیک نوې ناسته رامینځته کوي. د ساتلو وقفه پدې معنی ده چې د یوې ټاکلې مودې لپاره د ناستې ساتل او د ناوخته معلوماتو لپاره اجازه ورکوي چې د ناستې د غیر فعالیت دورې څخه هاخوا غزیږي مګر بیا هم ضمیمه کیدی شي. برسیره پردې، د ادغام په پایله کې د نوې ناستې پیل او پای د لومړني او وروستي نیټې / وخت ټاپه سره مطابقت لري.

راځئ چې د شمیرنې میتود څخه یو څو ننوتنې وګورو ترڅو وګورو چې ناستې څنګه کار کوي (جدول 5.1).

کله چې ریکارډونه راشي، موږ د ورته کیلي سره موجوده ناستې ګورو، د پای وخت د اوسني نیټې / وخت سټیمپ څخه کم دی - د غیر فعالیت وقفه، او د پیل وخت د اوسني نیټې / وخت سټیمپ + غیر فعالیت وقف څخه ډیر. د دې په پام کې نیولو سره، د میز څخه څلور ننوتل. 5.1 په لاندې ډول په یوه ناسته کې یوځای شوي.

1. ریکارډ 1 لومړی راځي، نو د پیل وخت د پای وخت سره برابر دی او 00:00:00 دی.

2. بیا، د ننوتلو 2 راځي، او موږ د غونډو په لټه کې یو چې د 23:59:55 څخه مخکې پای ته نه رسیږي او د 00:00:35 څخه وروسته پیل نشي. موږ ریکارډ 1 پیدا کوو او ناستې 1 او 2 سره یوځای کوو. موږ د ناستې د پیل وخت 1 (مخکې) او د 2 ناستې پای وخت (وروسته) اخلو، نو زموږ نوې ناسته په 00:00:00 پیل کیږي او په 00 پای ته رسیږي: 00:15

3. ریکارډ 3 راځي، موږ د 00:00:30 او 00:01:10 ترمنځ غونډو ته ګورو او هیڅ یې ونه موندل. د کلیدي 123-345-654،FFBE لپاره دویمه ناسته اضافه کړئ، په 00:00:50 پیل او پای ته رسیږي.

4. ریکارډ 4 راځي او موږ د 23:59:45 او 00:00:25 ترمنځ غونډو په لټه کې یو. دا ځل دواړه ناستې 1 او 2 موندل شوي. ټولې درې ناستې په یوه کې یوځای شوي، د پیل وخت 00:00:00 او د پای وخت 00:00:15 سره.

د هغه څه څخه چې پدې برخه کې تشریح شوي، دا د لاندې مهمو نښو یادولو ارزښت لري:

• سیشنونه د ثابت اندازې کړکۍ ندي. د ناستې موده د یوې ټاکلې مودې دننه د فعالیت لخوا ټاکل کیږي؛
• په ډیټا کې د نیټې / وخت ټاپهونه ټاکي چې ایا پیښه په موجوده ناستې کې راځي یا د غیر فعال دورې په جریان کې.

بیا به موږ د کړکۍ بل ډول په اړه بحث وکړو - "ټمبلینګ" کړکۍ.

کړکۍ "ټمبلینګ".

ټمبلینګ کړکۍ هغه پیښې نیسي چې د یوې ټاکلې مودې په اوږدو کې پیښیږي. تصور وکړئ چې تاسو اړتیا لرئ په هرو 20 ثانیو کې د یو ځانګړي شرکت ټول سټاک لیږدونه ونیسئ ، نو تاسو د دې مودې په جریان کې ټولې پیښې راټول کړئ. د 20-ثانوي وقفې په پای کې، کړکۍ پورته کیږي او د 20-ثانوي مشاهدې وقفې ته ځي. 5.14 شکل دا وضعیت روښانه کوي.

لکه څنګه چې تاسو لیدلی شئ، په تیرو 20 ثانیو کې ترلاسه شوي ټولې پیښې په کړکۍ کې شاملې دي. د دې مودې په پای کې ، یوه نوې کړکۍ رامینځته کیږي.

5.6 لیست کول کوډ ښیې چې په هر 20 ثانیو کې د سټاک لیږدونو د نیولو لپاره د ټمبلینګ وینډوز کارول ښیې (په src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java کې موندل کیږي).

د TimeWindows.of میتود کال ته د دې کوچني بدلون سره، تاسو کولی شئ د ټمبلینګ کړکۍ وکاروئ. دا مثال د till() میتود نه بولي، نو د 24 ساعتونو اصلي ساتل وقفه به وکارول شي.

په نهایت کې ، دا وخت دی چې د کړکۍ وروستي اختیارونو ته لاړشئ - "هاپینګ" وینډوز.

سلایډ ("کود") کړکۍ

سلیډینګ/هپنګ کړکۍ د ټمبلینګ کړکیو سره ورته دي، مګر د لږ توپیر سره. سلایډینګ کړکۍ د وروستیو پیښو پروسس کولو لپاره د نوې کړکۍ رامینځته کولو دمخه د وخت وقفې پای ته انتظار نه کوي. دوی د کړکۍ دورې څخه لږ انتظار وقفې وروسته نوي حسابونه پیل کوي.

د ټمبلینګ او جمپ کولو کړکۍ ترمینځ د توپیرونو روښانه کولو لپاره ، راځئ چې د سټاک تبادلې معاملو شمیرلو مثال ته راستون شو. زموږ هدف لاهم د معاملو شمیره ده، مګر موږ نه غواړو د کاونټر تازه کولو دمخه د ټول وخت انتظار وکړو. پرځای یې، موږ به کاونټر په لنډو وقفو کې تازه کړو. د مثال په توګه، موږ به بیا هم په هرو 20 ثانیو کې د معاملو شمیره حساب کړو، مګر کاونټر په هر 5 ثانیو کې تازه کړئ، لکه څنګه چې په انځور کې ښودل شوي. 5.15. په دې حالت کې، موږ د درې پایلې کړکۍ سره پای ته ورسوو د ډیټا ډیټا سره.

د 5.7 لیست کول د سلیډینګ وینډوز تعریف کولو لپاره کوډ ښیې (په src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java کې موندل شوي).

د ټمبلینګ کړکۍ د AdvanceBy() میتود ته د زنګ په اضافه کولو سره د هپ کولو کړکۍ ته بدل کیدی شي. په ښودل شوي مثال کې، د سپما وقفه 15 دقیقې ده.

تاسو پدې برخه کې ولیدل چې څنګه د وخت وینډوز ته د راټولولو پایلې محدود کړئ. په ځانګړې توګه، زه غواړم چې تاسو د دې برخې څخه لاندې درې شیان په یاد ولرئ:

• د سیشن وینډوز اندازه د وخت مودې پورې محدود نه ده، مګر د کاروونکي فعالیت لخوا محدوده ده؛
• "ټمبلینګ" کړکۍ د یوې ټاکلې مودې په اوږدو کې د پیښو عمومي کتنه وړاندې کوي؛
• د کود کولو کړکۍ موده ټاکل شوې ده، مګر دوی په مکرر ډول تازه کیږي او کیدای شي په ټولو کړکیو کې د ورپیښې داخلې ولري.

بیا، موږ به زده کړو چې څنګه د پیوستون لپاره KTable بیرته KStream ته بدل کړو.

5.3.3. د KStream او KTable شیانو سره نښلول

په څلورم څپرکي کې، موږ د دوو KStream شیانو سره د نښلولو په اړه بحث وکړ. اوس موږ باید زده کړو چې څنګه KTable او KStream سره وصل کړو. دا ممکن د لاندې ساده دلیل لپاره اړین وي. KStream د ریکارډونو یو جریان دی، او KTable د ریکارډ تازه کولو جریان دی، مګر ځینې وختونه تاسو غواړئ د KTable څخه د تازه معلوماتو په کارولو سره د ریکارډ جریان کې اضافي شرایط اضافه کړئ.

راځئ چې د سټاک ایکسچینج لیږدونو شمیر په اړه معلومات واخلو او د اړوندو صنعتونو لپاره د سټاک ایکسچینج خبرونو سره یوځای کړو. دلته هغه څه دي چې تاسو ورته اړتیا لرئ د دې لاسته راوړلو لپاره هغه کوډ ورکړئ چې تاسو دمخه لرئ.

1. د KTable څیز د سټاک لیږدونو شمیر په اړه ډیټا سره په KStream کې بدل کړئ ، وروسته د کیلي سره د کیلي ځای په ځای کړئ چې د دې سټاک سمبول سره ورته د صنعت سکتور په ګوته کوي.
2. د KTable څیز جوړ کړئ چې د یوې موضوع څخه ډاټا د سټاک ایکسچینج خبرونو سره لوستل کیږي. دا نوی KTable به د صنعت سکتور لخوا طبقه بندي شي.
3. د صنعت سکټور لخوا د سټاک تبادلې لیږدونو شمیر په اړه معلوماتو سره د خبرونو تازه معلومات وصل کړئ.

اوس راځئ چې وګورو چې دا عمل څنګه پلي کیږي.

KTable KStream ته واړوئ

KTable ته KStream بدلولو لپاره تاسو اړتیا لرئ لاندې کارونه وکړئ.

1. د KTable.toStream() میتود ته زنګ ووهئ.
2. د KStream.map میتود په زنګ وهلو سره، کیلي د صنعت نوم سره بدل کړئ، او بیا د کړکۍ مثال څخه د TransactionSummary څیز بیرته ترلاسه کړئ.

موږ به دا عملیات په لاندې ډول سره یو ځای کړو (کوډ په فایل کې موندل کیدی شي src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (5.8 لیست کول).

ځکه چې موږ د KStream.map عملیات ترسره کوو، د KStream بیرته راستنیدنه په اتوماتيک ډول بیا ویشل کیږي کله چې دا په اړیکه کې کارول کیږي.

موږ د تبادلې پروسه بشپړه کړې، بیا موږ اړتیا لرو چې د سټاک خبرونو لوستلو لپاره د KTable څیز جوړ کړو.

د سټاک خبرونو لپاره د KTable جوړول

خوشبختانه، د KTable څیز رامینځته کول یوازې د کوډ یوه کرښه اخلي (کوډ په src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java کې موندل کیدی شي) (5.9 لیست کول).

دا د یادولو وړ ده چې د سیرډ توکي مشخص کولو ته اړتیا نلري، ځکه چې سیرډز په ترتیباتو کې کارول کیږي. همدارنګه، د ابتدايي شمیرنې په کارولو سره، جدول په پیل کې د ریکارډونو څخه ډک شوی.

اوس موږ کولی شو وروستي ګام ته لاړ شو - پیوستون.

د لیږد شمیرې ډیټا سره د خبرونو تازه معلومات نښلول

د اړیکو جوړول ستونزمن نه دي. موږ به د کیڼ اړخ شمولیت وکاروو په هغه صورت کې چې د اړوند صنعت لپاره د سټاک خبرونه شتون نلري (ضروري کوډ په فایل کې موندل کیدی شي src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (لیست 5.10).

دا لیفٹ جوین آپریټر خورا ساده دی. په څلورم څپرکي کې د یوځای کیدو په څیر، د وینډوز طریقه نه کارول کیږي ځکه چې کله چې د KStream-KTable یوځای کول ترسره کوي، په KTable کې د هرې کیلي لپاره یوازې یوه ننوتل شتون لري. دا ډول اړیکه په وخت کې محدوده نه ده: ریکارډ یا په KTable کې وي یا غیر حاضر وي. اصلي پایله: د KTable څیزونو په کارولو سره تاسو کولی شئ د لږ وخت تازه شوي حوالې ډیټا سره KStream بډایه کړئ.

اوس به موږ د KStream څخه د پیښو بډایه کولو لپاره خورا مؤثره لاره وګورو.

5.3.4. GlobalKTable توکي

لکه څنګه چې تاسو لیدلی شئ ، د پیښې جریان بډایه کولو یا دوی ته شرایط اضافه کولو اړتیا شتون لري. په څلورم څپرکي کې تاسو د KStream د دوو شیانو ترمنځ اړیکې ولیدلې، او په تیره برخه کې تاسو د KStream او KTable ترمنځ اړیکه ولیدله. په دې ټولو قضیو کې ، دا اړینه ده چې د ډیټا جریان بیا تقسیم کړئ کله چې نوي ډول یا ارزښت ته کیلي نقشه کړئ. ځینې ​​​​وختونه بیا تقسیم کول په ښکاره ډول ترسره کیږي، او ځینې وختونه د کافکا سټریمونه دا په اوتومات ډول ترسره کوي. بیا ویشل اړین دي ځکه چې کیلي بدل شوي او ریکارډونه باید په نویو برخو کې پای ته ورسیږي، که نه نو پیوستون به ناشونی وي (دا په 4 څپرکي کې، د 4 فرعي برخې "د معلوماتو بیا ویشل" برخه کې بحث شوی).

بیا ویشل لګښت لري

بیا ویشل لګښتونو ته اړتیا لري - د منځنیو موضوعاتو رامینځته کولو لپاره اضافي سرچینې لګښتونه، په بله موضوع کې د نقل ډاټا ذخیره کول؛ دا د دې موضوع څخه د لیکلو او لوستلو له امله د ځنډ د زیاتوالي معنی هم لري. برسیره پردې، که تاسو اړتیا لرئ چې د یو څخه ډیرو اړخونو یا اړخونو سره یوځای شئ، تاسو باید د یوځای کیدو سلسله جوړه کړئ، ریکارډونه د نوي کیلي سره نقشه کړئ، او د بیا ویشلو پروسه بیا پرمخ بوځي.

د کوچنیو ډیټاسیټونو سره نښلول

په ځینو مواردو کې، د وصل کولو لپاره د حوالې ډیټا حجم نسبتا کوچنی دی، نو د دې بشپړ کاپي کولی شي په اسانۍ سره په هر نوډ کې ځایی وي. د دې په څیر حالتونو لپاره، کافکا جریان د GlobalKTable ټولګي چمتو کوي.

د GlobalKTable مثالونه بې ساري دي ځکه چې غوښتنلیک هر نوډ ته ټول معلومات نقلوي. او څنګه چې ټول معلومات په هر نوډ کې شتون لري ، نو اړتیا نشته چې د پیښې جریان د حوالې ډیټا کیلي لخوا ویشل شي ترڅو دا ټولو برخو ته شتون ولري. تاسو کولی شئ د GlobalKTable شیانو په کارولو سره د کیلي پرته یوځای کیدل هم کولی شئ. راځئ چې د دې ځانګړتیا د ښودلو لپاره یو پخوانی مثال ته لاړ شو.

د KStream څیزونه د GlobalKTable شیانو سره نښلول

په 5.3.2 فرعي برخه کې، موږ د پیرودونکو لخوا د تبادلې د راکړې ورکړې د کړکۍ راټولول ترسره کړل. د دې مجموعې پایلې داسې ښکاري:

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

پداسې حال کې چې دا پایلې هدف ته رسیدلي، دا به ډیر ګټور وي که چیرې د پیرودونکي نوم او د شرکت بشپړ نوم هم ښودل شوی وای. د پیرودونکي نوم او شرکت نوم اضافه کولو لپاره، تاسو کولی شئ نورمال شمولیت ترسره کړئ، مګر تاسو به دوه کلیدي نقشه کولو او بیا ویشلو ته اړتیا ولرئ. د GlobalKTable سره تاسو کولی شئ د داسې عملیاتو لګښت څخه مخنیوی وکړئ.

د دې کولو لپاره، موږ به د لیست کولو 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. دا جدول د کافکا سټریمونو 1.0.0 نسخه سره د ارتباط وړتیاوې منعکس کوي؛ یو څه ممکن په راتلونکو خپرونو کې بدلون ومومي.

د شیانو د سمولو لپاره ، راځئ چې اساسات بیا تکرار کړو: تاسو کولی شئ د ځایی حالت په کارولو سره د پیښې جریان (KStream) وصل کړئ او جریانونه (KTable) تازه کړئ. په بدیل سره، که د حوالې ډاټا اندازه خورا لوی نه وي، تاسو کولی شئ د GlobalKTable څیز وکاروئ. GlobalKTables ټولې برخې د کافکا سټریمز غوښتنلیک نوډ ته نقلوي، دا ډاډه کوي چې ټول معلومات شتون لري پرته لدې چې کیلي د کومې برخې سره مطابقت ولري.

بیا به موږ د کافکا سټیمز فیچر وګورو، د کوم څخه مننه چې موږ کولی شو د کافکا موضوع څخه د معلوماتو مصرف کولو پرته د دولت بدلونونه وګورو.

5.3.5. د پوښتنې وړ حالت

موږ لا دمخه ډیری عملیات ترسره کړي دي چې ریاست پکې شامل دي او تل پایلې کنسول ته (د پراختیایي موخو لپاره) یا موضوع ته لیکو (د تولید اهدافو لپاره). کله چې یوې موضوع ته پایلې ولیکئ، نو تاسو باید د دوی د لیدلو لپاره د کافکا پیرودونکي وکاروئ.

د دې موضوعاتو څخه د معلوماتو لوستل د مادي نظرونو یو ډول ګڼل کیدی شي. زموږ د موخو لپاره، موږ کولی شو د ویکیپیډیا څخه د مادي لید تعریف وکاروو: "... یو فزیکي ډیټابیس څیز چې د پوښتنې پایلې لري. د مثال په توګه، دا کیدای شي د لرې پرتو معلوماتو محلي کاپي وي، یا د قطارونو او/یا د جدول د کالمونو فرعي سیټ یا د پایلو سره یوځای شي، یا د لنډیز جدول د راټولولو له لارې ترلاسه شوی وي" (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