streamalism.org

1. הקדמה

במדריך זה נלמד על Hazelcast Jet. זהו מנוע לעיבוד נתונים מבוזר המסופק על ידי Hazelcast, Inc והוא בנוי על גבי IMDG של Hazelcast.

אם אתה רוצה ללמוד על IMDG של Hazelcast, הנה מאמר לתחילת העבודה.

2. מה זה Jet Hazelcast?

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

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

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

להלן מקרי שימוש אופייניים ל- Hazelcast Jet:

• עיבוד זרם בזמן אמת
• עיבוד אצווה מהיר
• עיבוד Java 8 Streams בצורה מבוזרת
• עיבוד נתונים במיקרו-שירותים

3. התקנה

כדי להגדיר את Hazelcast Jet בסביבתנו, אנחנו רק צריכים להוסיף תלות אחת של Maven שלנו pom.xml.

כך אנו עושים זאת:

 com.hazelcast.jet hazelcast-jet 4.2 

הכללת תלות זו תוריד קובץ צנצנת של 10 מגהבייט המספק לנו את כל התשתית הדרושה לנו לבניית צינור עיבוד נתונים מבוזר.

הגרסה האחרונה של Hazelcast Jet נמצאת כאן.

4. יישום לדוגמא

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

4.1. הצינור

צינור מהווה את המבנה הבסיסי ליישום Jet. עיבוד בצינור מבצע את השלבים הבאים:

• לקרוא נתונים ממקור
• להפוך את הנתונים
• לכתוב נתונים לכיור

עבור היישום שלנו, הצינור יקרא ממופץ רשימה, ליישם את השינוי של קיבוץ וצבירה ולבסוף לכתוב להפצה מַפָּה.

כך אנו כותבים את הצינור שלנו:

צינור פרטי createPipeLine () {Pipeline p = Pipeline.create (); p.readFrom (Sources.list (LIST_NAME)) .flatMap (word -> traverseArray (word.toLowerCase (). split ("\ W +"))). filter (word ->! word.isEmpty ()) .groupingKey (wholeItem ()) .aggregate (counting ()) .writeTo (Sinks.map (MAP_NAME)); החזר p; }

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

לבסוף, אנו מקבצים את המילים, מצטברים אותן וכותבים את התוצאות ל- a מַפָּה.

4.2. העבודה

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

כך אנו כותבים א countWord פונקציה המקבלת פרמטרים ומחזירה את הספירה:

ספירה ארוכה ציבורית (משפטים ברשימה, מחרוזת) {ספירה ארוכה = 0; סילון JetInstance = Jet.newJetInstance (); נסה את {List textList = jet.getList (LIST_NAME); textList.addAll (משפטים); צינור p = createPipeLine (); jet.newJob (p) .join (); ספירת מפות = jet.getMap (MAP_NAME); count = counts.get (מילה); } סוף סוף {Jet.shutdownAll (); } ספירת החזרות; }

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

לאחר מכן אנו מגישים עבודה באמצעות הצינור שבנינו לעיל. השיטה עבודה חדשה() מחזירה עבודת הפעלה שמופעלת על ידי Jet בצורה לא סינכרונית. ה לְהִצְטַרֵף שיטה ממתין לביצוע העבודה וזורק יוצא מן הכלל אם העבודה הושלמה עם שגיאה.

לאחר סיום העבודה התוצאות מתקבלות בהפצה מַפָּה, כפי שהגדרנו בצנרת שלנו. אז, אנחנו מקבלים את מַפָּה ממופע Jet וקבלו את ספירת המילה כנגדו.

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

הנה מבחן יחידה הבודק את הקוד שכתבנו עבור Jet:

@Test הציבור בטל כאשר GivenSentencesAndWord_ThenReturnCountOfWord () {רשימת משפטים = ArrayList חדש (); sentences.add ("השנייה הראשונה הייתה בסדר, אבל השנייה השנייה הייתה קשה."); WordCounter wordCounter = מונה חדש (); count countSecond ארוך = wordCounter.countWord (משפטים, "שנייה"); assertEquals (3, countSecond); }

5. מסקנה

במאמר זה למדנו על Hazelcast Jet. למידע נוסף אודותיו ותכונותיו, עיין במדריך.

כרגיל, את הקוד לדוגמאות המשמשות במאמר זה ניתן למצוא באתר Github.