הספר "זרמי קפקא בפעולה. יישומים ומיקרו-שירותים לעבודה בזמן אמת"

שלום תושבי חברו! ספר זה מתאים לכל מפתח שרוצה להבין עיבוד שרשור. הבנת תכנות מבוזר תעזור לך להבין טוב יותר את קפקא וקפקא זרמים. זה יהיה נחמד להכיר את מסגרת קפקא עצמה, אבל זה לא הכרחי: אני אגיד לך כל מה שאתה צריך. מפתחי קפקא מנוסים וטירונים כאחד ילמדו כיצד ליצור יישומי עיבוד זרמים מעניינים באמצעות ספריית Kafka Streams בספר זה. מפתחי Java בינוניים ומתקדמים שכבר מכירים מושגים כמו סדרה ילמדו ליישם את כישוריהם ליצירת יישומי Kafka Streams. קוד המקור של הספר כתוב ב-Java 8 ועושה שימוש משמעותי בתחביר הביטוי של Java 8 lambda, כך שידע לעבוד עם פונקציות למבדה (אפילו בשפת תכנות אחרת) יהיה שימושי.

קטע. 5.3. פעולות צבירה וחלון

בחלק זה, נעבור לחקור את החלקים המבטיחים ביותר של קפקא זרמים. עד כה כיסינו את ההיבטים הבאים של קפקא זרמי:

• יצירת טופולוגיית עיבוד;
• שימוש במצב באפליקציות סטרימינג;
• ביצוע חיבורי זרם נתונים;
• הבדלים בין זרמי אירועים (KStream) לזרמי עדכון (KTable).

בדוגמאות הבאות נביא את כל המרכיבים הללו יחד. תלמד גם על חלונות, תכונה נהדרת נוספת של יישומי סטרימינג. הדוגמה הראשונה שלנו תהיה צבירה פשוטה.

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) .

אתחול זה מכיל משתנה קבוע בתור. זהו אובייקט מותאם אישית המהווה מתאם עבור java.util.TreeSet המשמש למעקב אחר תוצאות N העליונות בסדר יורד של המניות הנסחרות.

כבר ראית את הקריאות groupBy ו-mapValues, אז לא ניכנס לאלו (אנחנו קוראים לשיטת KTable.toStream מכיוון ששיטת KTable.print הוצאה משימוש). אבל עדיין לא ראית את גרסת KTable של aggregate() אז נקדיש זמן לדיון על זה.

כפי שאתה זוכר, מה שמייחד את KTable הוא שרשומות עם אותם מפתחות נחשבות לעדכונים. KTable מחליף את הערך הישן בחדש. צבירה מתרחשת באופן דומה: הרשומות האחרונות עם אותו מפתח מצטברים. כאשר רשומה מגיעה, היא מתווספת למופע המחלקה FixedSizePriorityQueue באמצעות adder (פרמטר שני בקריאת השיטה המצטברת), אך אם רשומה אחרת כבר קיימת עם אותו מפתח, אז הרשומה הישנה מוסרת באמצעות מחסיר (פרמטר שלישי ב- קריאת השיטה המצטברת).

כל זה אומר שהאגרגטור שלנו, FixedSizePriorityQueue, אינו צובר את כל הערכים במפתח אחד, אלא מאחסן סכום נע של הכמויות של N סוגי המניות הנסחרות ביותר. כל ערך נכנס מכיל את המספר הכולל של המניות שנמכרו עד כה. KTable תיתן לך מידע לגבי המניות של החברות הנסחרות כיום ביותר, מבלי להידרש לצבירה מתגלגלת של כל עדכון.

למדנו לעשות שני דברים חשובים:

• קבץ ערכים ב-KTable על ידי מפתח משותף;
• לבצע פעולות שימושיות כגון אוסף וצבירה על ערכים מקובצים אלה.

לדעת כיצד לבצע פעולות אלו חשוב להבין את המשמעות של הנתונים העוברים דרך אפליקציית Kafka Streams ולהבין איזה מידע הוא נושא.

ריכזנו גם כמה ממושגי המפתח שנדונו קודם לכן בספר זה. בפרק 4, דנו עד כמה המדינה המקומית סובלנית לתקלות היא חשובה עבור אפליקציית סטרימינג. הדוגמה הראשונה בפרק זה הדגימה מדוע המדינה המקומית כל כך חשובה - היא מאפשרת לך לעקוב אחר המידע שכבר ראית. גישה מקומית מונעת עיכובים ברשת, מה שהופך את האפליקציה ליותר ביצועית ועמידה בפני שגיאות.

בעת ביצוע פעולת אוסף או צבירה כלשהי, עליך לציין את שם חנות המדינה. פעולות האוסף והצבירה מחזירות מופע של KTable, וה-KTable משתמש באחסון מצב כדי להחליף תוצאות ישנות בחדשות. כפי שראית, לא כל העדכונים נשלחים בצנרת, וזה חשוב מכיוון שפעולות צבירה נועדו לייצר מידע מסכם. אם לא תחיל את המדינה המקומית, KTable תעביר את כל תוצאות הצבירה והאוסף.

לאחר מכן, נבחן ביצוע פעולות כגון צבירה בתוך פרק זמן מסוים - מה שנקרא פעולות חלונות.

5.3.2. פעולות חלונות

בסעיף הקודם, הצגנו קונבולולוציה וצבירה של החלקה. האפליקציה ביצעה רול-אפ מתמשך של מכירות המניות ולאחריה צבירת חמש המניות הנסחרות ביותר בבורסה.

לפעמים יש צורך בצבירה רציפה כזו של תוצאות. ולפעמים אתה צריך לבצע פעולות רק על פני פרק זמן נתון. לדוגמה, חשב כמה עסקאות חליפין בוצעו עם מניות של חברה מסוימת ב-10 הדקות האחרונות. או כמה משתמשים לחצו על באנר פרסומי חדש ב-15 הדקות האחרונות. יישום עשוי לבצע פעולות כאלה מספר פעמים, אך עם תוצאות החלות רק על פרקי זמן מוגדרים (חלונות זמן).

ספירת עסקאות חליפין לפי קונה

בדוגמה הבאה, נעקוב אחר עסקאות מניות בין מספר סוחרים - ארגונים גדולים או אנשי כספים בודדים חכמים.

ישנן שתי סיבות אפשריות למעקב זה. אחד מהם הוא הצורך לדעת מה מובילי שוק קונים/מוכרים. אם השחקנים הגדולים והמשקיעים המתוחכמים האלה רואים הזדמנות, זה הגיוני לעקוב אחר האסטרטגיה שלהם. הסיבה השנייה היא הרצון לזהות כל סימן אפשרי של סחר בפנים לא חוקי. כדי לעשות זאת, תצטרך לנתח את המתאם של עליות מכירות גדולות עם הודעות לעיתונות חשובות.

מעקב כזה מורכב מהשלבים הבאים:

• יצירת זרם לקריאה מתוך נושא עסקאות מניות;
• קיבוץ רשומות נכנסות לפי מזהה קונה וסמל המניה. קריאה למתודה 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(fifteenMinutes)) אנו יוצרים חלון הפעלה עם מרווח חוסר פעילות של 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, אתה יכול להשתמש בחלון נפילה. דוגמה זו אינה קוראת למתודה () עד, ולכן ישמש מרווח השמירה המוגדר כברירת מחדל של 24 שעות.

לבסוף, הגיע הזמן לעבור לאחרון מבין אפשרויות החלונות - חלונות "קפיצה".

חלונות הזזה ("קופצים")

חלונות הזזה/דילוגים דומים לחלונות נפולים, אך עם הבדל קל. חלונות הזזה אל תחכו לסוף מרווח הזמן לפני יצירת חלון חדש לעיבוד אירועים אחרונים. הם מתחילים חישובים חדשים לאחר מרווח המתנה הקטן ממשך החלון.

כדי להמחיש את ההבדלים בין חלונות נפילה לקפיצה, נחזור לדוגמא של ספירת עסקאות בבורסה. המטרה שלנו היא עדיין לספור את מספר העסקאות, אבל אנחנו לא רוצים לחכות את כל הזמן לפני עדכון המונה. במקום זאת, נעדכן את המונה במרווחים קצרים יותר. לדוגמה, עדיין נספור את מספר העסקאות כל 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) (רישום 5.9).

ראוי לציין כי אין צורך לציין אובייקטים של Serde, מכיוון שהמחרוזת Serdes משמשת בהגדרות. כמו כן, על ידי שימוש בספירה המוקדמת ביותר, הטבלה מתמלאת ברשומות ממש בהתחלה.

כעת נוכל לעבור לשלב האחרון – חיבור.

חיבור עדכוני חדשות עם נתוני ספירת עסקאות

יצירת חיבור אינה קשה. נשתמש בהצטרפות שמאלה למקרה שאין חדשות מלאי לענף הרלוונטי (את הקוד הדרוש ניתן למצוא בקובץ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (רישום 5.10).

אופרטור leftJoin זה די פשוט. בניגוד ל-joins בפרק 4, לא נעשה שימוש בשיטת JoinWindow מכיוון שכאשר מבצעים צירוף KStream-KTable, יש רק כניסה אחת ב-KTable לכל מפתח. חיבור כזה אינו מוגבל בזמן: הרשומה נמצאת או ב-KTable או נעדרת. המסקנה העיקרית: באמצעות אובייקטי KTable אתה יכול להעשיר את KStream בנתוני התייחסות המתעדכנים בתדירות נמוכה יותר.

כעת נבחן דרך יעילה יותר להעשיר אירועים מ-KStream.

5.3.4. אובייקטים GlobalKTable

כפי שאתה יכול לראות, יש צורך להעשיר את זרמי האירועים או להוסיף להם הקשר. בפרק 4 ראית את החיבורים בין שני אובייקטים של KStream, ובסעיף הקודם ראית את הקשר בין KStream ל-KTable. בכל המקרים הללו, יש צורך לחלק מחדש את זרם הנתונים בעת מיפוי המפתחות לסוג או לערך חדש. לפעמים החלוקה מחדש מתבצעת במפורש, ולפעמים קפקא סטרים עושה זאת אוטומטית. חלוקה מחדש נחוצה מכיוון שהמפתחות השתנו והרשומות חייבות להגיע למקטעים חדשים, אחרת החיבור יהיה בלתי אפשרי (הדבר נדון בפרק 4, בסעיף "חלוקה מחדש של נתונים" בתת-סעיף 4.2.4).

לחלוקה מחדש יש עלות

חלוקה מחדש מצריכה עלויות - עלויות משאבים נוספות ליצירת נושאי ביניים, אחסון נתונים כפולים בנושא אחר; המשמעות היא גם חביון מוגבר עקב כתיבה וקריאה מהנושא הזה. בנוסף, אם אתה צריך להצטרף על פני יותר מאספקט או מימד אחד, עליך לשרשר את החיבורים, למפות את הרשומות עם מפתחות חדשים ולהפעיל שוב את תהליך החלוקה מחדש.

חיבור למערכי נתונים קטנים יותר

במקרים מסוימים, נפח נתוני ההתייחסות לחיבור קטן יחסית, כך שעותקים שלמים שלהם יכולים להתאים בקלות מקומית לכל צומת. עבור מצבים כאלה, קפקא סטרים מספק את הכיתה GlobalKTable.

מופעי GlobalKTable הם ייחודיים מכיוון שהאפליקציה משכפלת את כל הנתונים לכל אחד מהצמתים. ומכיוון שכל הנתונים קיימים בכל צומת, אין צורך לחלק את זרם האירועים לפי מפתח נתוני הפניה כך שהוא זמין לכל המחיצות. אתה יכול גם ליצור חיבורים ללא מפתח באמצעות אובייקטים של 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 מופשט לאובייקט פונקציה במקום ביטוי למבדה מוטבע למען הקריאה.

השלב הבא הוא להכריז על שני מופעים של 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 של Kafka Streams; משהו עשוי להשתנות במהדורות עתידיות.

כדי לסכם את הדברים, בואו נסכם את היסודות: אתה יכול לחבר זרמים של אירועים (KStream) ולעדכן זרמים (KTable) באמצעות מצב מקומי. לחלופין, אם גודל נתוני ההפניה אינו גדול מדי, אתה יכול להשתמש באובייקט GlobalKTable. GlobalKTables משכפלים את כל המחיצות לכל צומת אפליקציית Kafka Streams, ומבטיחים שכל הנתונים זמינים ללא קשר לאיזו מחיצה המפתח מתאים.

בשלב הבא נראה את תכונת קפקא זרמי, שבזכותה נוכל לצפות בשינויי מצב מבלי לצרוך נתונים מנושא קפקא.

5.3.5. מצב שניתן לשאול

כבר ביצענו מספר פעולות הכוללות מצב ותמיד פלטנו את התוצאות לקונסולה (למטרות פיתוח) או כתבנו אותן לנושא (למטרות ייצור). בעת כתיבת תוצאות לנושא, עליך להשתמש בצרכן קפקא כדי לצפות בהן.

קריאת נתונים מהנושאים הללו יכולה להיחשב כסוג של השקפות ממומשות. לענייננו, אנו יכולים להשתמש בהגדרה של תצוגה ממומשת מויקיפדיה: "...אובייקט של מסד נתונים פיזי המכיל תוצאות של שאילתה. לדוגמה, זה יכול להיות עותק מקומי של נתונים מרוחקים, או תת-קבוצה של השורות ו/או העמודות של טבלה או תוצאות הצטרפות, או טבלת סיכום שהושגה באמצעות צבירה" (https://en.wikipedia.org/wiki /Materialized_view).

Kafka Streams גם מאפשר לך להריץ שאילתות אינטראקטיביות בחנויות המדינה, מה שמאפשר לך לקרוא ישירות את הדעות שהתממשו. חשוב לציין שהשאילתה לחנות המדינה היא פעולה לקריאה בלבד. זה מבטיח שלא תצטרך לדאוג להפיכת המצב בטעות לבלתי עקבי בזמן שהאפליקציה שלך מעבדת נתונים.

היכולת לבצע שאילתות ישירות בחנויות המדינה חשובה. זה אומר שאתה יכול ליצור יישומי לוח מחוונים מבלי שתצטרך קודם להביא נתונים מצרכן קפקא. זה גם מגביר את היעילות של האפליקציה, בשל העובדה שאין צורך לכתוב נתונים שוב:

• הודות למקומם של הנתונים, ניתן לגשת אליהם במהירות;
• שכפול הנתונים נמחק, מכיוון שהוא לא נכתב לאחסון חיצוני.

הדבר העיקרי שאני רוצה שתזכור הוא שאתה יכול לשאול את המצב ישירות מתוך היישום שלך. אי אפשר להפריז בהזדמנויות שזה נותן לך. במקום לצרוך נתונים מקפקא ולאחסן רשומות במסד נתונים עבור האפליקציה, ניתן לבצע שאילתות בחנויות מצב עם אותה תוצאה. שאילתות ישירות לחנויות המדינה אומרות פחות קוד (ללא צרכן) ופחות תוכנה (אין צורך בטבלת מסד נתונים כדי לאחסן את התוצאות).

כיסינו לא מעט מהשטח בפרק הזה, אז נעזוב את הדיון שלנו בשאילתות אינטראקטיביות נגד חנויות ממלכתיות לעת עתה. אבל אל דאגה: בפרק 9, ניצור יישום לוח מחוונים פשוט עם שאילתות אינטראקטיביות. הוא ישתמש בכמה מהדוגמאות מפרקים זה ומפרקים קודמים כדי להדגים שאילתות אינטראקטיביות וכיצד ניתן להוסיף אותן ליישומי Kafka Streams.

תקציר

• אובייקטי KStream מייצגים זרמים של אירועים, הדומים להוספות למסד נתונים. אובייקטי KTable מייצגים זרמי עדכון, יותר כמו עדכונים למסד נתונים. גודל האובייקט KTable לא גדל, רשומות ישנות מוחלפות בחדשות.
• אובייקטי KTable נדרשים לפעולות צבירה.
• באמצעות פעולות חלונות, אתה יכול לפצל נתונים מצטברים לדליי זמן.
• הודות לאובייקטים של GlobalKTable, אתה יכול לגשת לנתוני התייחסות בכל מקום באפליקציה, ללא קשר לחלוקה.
• חיבורים בין אובייקטי KStream, KTable ו-GlobalKTable אפשריים.

עד כה, התמקדנו בבניית יישומי Kafka Streams באמצעות KStream DSL ברמה גבוהה. למרות שהגישה ברמה הגבוהה מאפשרת לך ליצור תוכניות מסודרות ותמציתיות, השימוש בה מייצג פשרה. עבודה עם DSL KStream פירושה הגדלת התמציתיות של הקוד שלך על ידי הפחתת מידת השליטה. בפרק הבא, נסתכל על ה-API של צומת המטפל ברמה נמוכה וננסה פשרות אחרות. התוכניות יהיו ארוכות יותר ממה שהן היו בעבר, אבל נוכל ליצור כמעט כל צומת מטפל שאולי נצטרך.

→ פרטים נוספים על הספר ניתן למצוא בכתובת אתר האינטרנט של ההוצאה לאור

→ להברוז'תילי 25% הנחה באמצעות קופון - נחלי קפקא

← עם התשלום עבור גרסת הנייר של הספר ישלח ספר אלקטרוני במייל.

מקור: www.habr.com