שלום לכולם!

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

NiFi אתר לאתר (S2S) היא דרך מאובטחת הניתנת להגדרה בקלות להעברת נתונים בין מופעי NiFi. איך S2S עובד, ראה תיעוד וחשוב לא לשכוח להגדיר את מופע ה-NiFi כדי לאפשר S2S, ראה כאן.

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

1. דחוף. ממופע הלקוח, הנתונים נשלחים באמצעות קבוצת תהליכים מרוחקת (RPG). במופע השרת, הנתונים מתקבלים באמצעות יציאת הקלט
2. משוך. השרת מקבל נתונים באמצעות RPG, הלקוח שולח באמצעות יציאת פלט.

זרימת ההפצה מאוחסנת ב- Apache Registry.

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

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

אבל ככל ש-N גדל, הבעיות נעשות רבות יותר:

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

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

1. השתמש ב-MiNiFi במקום ב-NiFi
2. NiFi CLI
3. NiPyAPI

שימוש ב-MiNiFi

Apache MiNiFy - פרויקט משנה של Apache NiFi. MiNiFy הוא סוכן קומפקטי שמשתמש באותם מעבדים כמו NiFi, ומאפשר לך ליצור את אותן זרימות כמו ב-NiFi. אופיו הקל משקל של הסוכן מושג, בין היתר, על ידי העובדה של-MiNiFy אין ממשק גרפי לתצורת זרימה. היעדר ממשק גרפי ב-MiNiFy אומר שיש צורך לפתור את בעיית העברת הזרימה ל-minifi. מכיוון ש-MiNiFy נמצא בשימוש פעיל ב-IOT, ישנם רכיבים רבים ותהליך אספקת הזרימה למופעי ה-Minifi הסופיים צריך להיות אוטומטי. משימה מוכרת, נכון?

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

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

1. ל-Minifi אין את כל המעבדים מ-nifi
2. גרסאות מעבד Minifi מפגרות אחרי גרסאות מעבד NiFi.

בזמן הכתיבה, הגרסה העדכנית ביותר של NiFi היא 1.9.2. גרסת מעבד ה-MiNiFi העדכנית ביותר היא 1.7.0. ניתן להוסיף מעבדים ל-MiNiFi, אך עקב פערי גרסאות בין מעבדי NiFi ו-MiNiFi, ייתכן שהדבר לא יעבוד.

NiFi CLI

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

הפעל את כלי השירות

./bin/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  /  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' to see a list of available commands, use tab to auto-complete.

על מנת שנוכל לטעון את הזרימה הנדרשת מהרישום, עלינו לדעת את המזהים של הדלי (מזהה דלי) והזרימה עצמה (מזהה זרימה). ניתן לקבל נתונים אלה דרך ה-cli או בממשק האינטרנט של הרישום של NiFi. בממשק האינטרנט זה נראה כך:

באמצעות CLI זה נעשה:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

אנו מתחילים לייבא קבוצת תהליכים מהרישום:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

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

קבוצת תהליכים שנוספה עם מעבדים שהופסקו, יש להפעיל אותם

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

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

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

מכיוון שיש לנו bash ואנחנו מוכנים ללכת עד הסוף, נמצא מוצא! אתה יכול להשתמש ב-NiFi API כדי לפתור בעיה זו. בוא נשתמש בשיטה הבאה, ניקח את המזהה מהדוגמאות למעלה (במקרה שלנו זה 7f522a13-016e-1000-e504-d5b15587f2f3). תיאור שיטות ה-API של NiFi כאן.

בגוף אתה צריך לעבור את JSON, כך:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

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

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

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

NiPyAPI

NiPyAPI היא ספריית Python לאינטראקציה עם מופעי NiFi. דף תיעוד מכיל את המידע הדרוש לעבודה עם הספרייה. התחלה מהירה מתוארת ב את הפרויקט ב-github.

הסקריפט שלנו להפעלת התצורה הוא תוכנית ב- Python. בואו נעבור לקידוד.
הגדרנו הגדרות לעבודה נוספת. נצטרך את הפרמטרים הבאים:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #путь до nifi-api инстанса, на котором разворачиваем process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #путь до nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #название registry, как будет называться в инстансе nifi
nipyapi.config.bucket_name = 'BucketName' #название bucket, из которого подтягиваем flow
nipyapi.config.flow_name = 'FlowName' #название flow, которое подтягиваем

לאחר מכן אכניס את שמות השיטות של ספרייה זו, המתוארות כאן.

חבר את הרישום למופע nifi באמצעות

nipyapi.versioning.create_registry_client

בשלב זה, תוכל גם להוסיף בדיקה שהרישום כבר התווסף למופע; לשם כך תוכל להשתמש בשיטה

nipyapi.versioning.list_registry_clients

אנו מוצאים את הדלי לחיפוש נוסף אחר זרימה בסל

nipyapi.versioning.get_registry_bucket

בעזרת הדלי שנמצא, אנו מחפשים זרימה

nipyapi.versioning.get_flow_in_bucket

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

nipyapi.canvas.list_all_process_groups

אנחנו יכולים להמשיך לחפש, למשל, לפי שם.

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

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

nipyapi.versioning.get_latest_flow_ver

קבוצת תהליכי פריס:

nipyapi.versioning.deploy_flow_version

אנו מפעילים את המעבדים:

nipyapi.canvas.schedule_process_group

בבלוק על CLI נכתב שהעברת נתונים לא מתאפשרת אוטומטית בקבוצת התהליך המרוחקת? בעת הטמעת הסקריפט, גם אני נתקלתי בבעיה זו. באותו זמן, לא הצלחתי להתחיל בהעברת נתונים באמצעות ה-API והחלטתי לכתוב למפתח ספריית NiPyAPI ולבקש עצה/עזרה. היזם הגיב לי, דנו בבעיה והוא כתב שהוא צריך זמן "לבדוק משהו". ואז, כמה ימים לאחר מכן, מגיע מכתב בו כתובה פונקציה בפייתון שפותרת את בעיית ההשקה שלי!!! באותו זמן, גרסת NiPyAPI הייתה 0.13.3 וכמובן, לא היה דבר כזה. אבל בגרסה 0.14.0, ששוחררה ממש לאחרונה, פונקציה זו כבר נכללה בספרייה. לִפְגוֹשׁ,

nipyapi.canvas.set_remote_process_group_transmission

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

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

זה הכל. תיארתי 3 גישות לאוטומציה של אספקת זרימה ב-NiFi, מלכודות שמפתח עלול להיתקל בהן, וסיפקתי קוד עבודה לאוטומציה של מסירה. אם אתה מתעניין בנושא הזה כמוני - לִכתוֹב!

מקור: www.habr.com