Apache NIFI - סקירה קצרה של תכונות בפועל

מבוא

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

משימה

צור 3 שירותים. הראשון שבהם הוא שירות עדכון מסד הנתונים. שירות זה, כאשר מגיעים נתונים חדשים ממערכת צד שלישי, מעדכן את הנתונים במסד הנתונים ויוצר קובץ בפורמט CSV להעברתם למערכת הבאה. נקודת הקצה של השירות השני נקראת - FTP Transport Service, שמקבל את הקובץ המועבר, מאמת אותו ומכניס אותו לאחסון קבצים באמצעות FTP. השירות השלישי, שירות העברת הנתונים לצרכן, פועל באופן אסינכרוני עם שני הראשונים. הוא מקבל בקשה ממערכת חיצונית של צד שלישי לקבל את הקובץ הנדון לעיל, לוקח את קובץ התגובה המוכנה, משנה אותו (מעדכן את ה-ID, התיאור, שדות linkToFile) ושולח את התגובה בצורה של הודעת SOAP. כלומר, התמונה הכוללת היא כדלקמן: שני השירותים הראשונים מתחילים את עבודתם רק כאשר הנתונים לעדכון הגיעו. השירות השלישי עובד ללא הרף כי ישנם צרכני מידע רבים, כ-1000 בקשות לנתונים בדקה. השירותים זמינים כל הזמן והמופעים שלהם ממוקמים בסביבות שונות, כגון בדיקה, הדגמה, טרום ייצור ופרוד. להלן תרשים של אופן הפעולה של שירותים אלה. הרשו לי להבהיר מיד שחלק מהפרטים פושטו כדי למנוע מורכבות מיותרת.

העמקה טכנית

כשתכננו פתרון לבעיה, החלטנו תחילה לעשות אפליקציות בג'אווה באמצעות ה-Spring framework, Nginx balancer, Postgres database ועוד דברים טכניים ולא כל כך טכניים. מכיוון שהזמן לפיתוח פתרון טכני אפשר לנו לשקול גישות אחרות לפתרון בעיה זו, מבטנו נפל על טכנולוגיית Apache NIFI, שהיא אופנתית בחוגים מסוימים. אני אגיד מיד שהטכנולוגיה הזו אפשרה לנו להבחין ב-3 השירותים האלה. מאמר זה יתאר את הפיתוח של שירות העברת קבצים ושירות העברת נתונים לצרכן, אך אם המאמר מועיל, אכתוב על השירות לעדכון נתונים במאגר.

מה זה

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

דוגמה

נחשבת דוגמה לאופן שבו ריבועים מתקשרים זה עם זה. הסכמה הכללית די פשוטה: אנו מקבלים בקשת HTTP (בתיאוריה, עם קובץ בגוף הבקשה. כדי להדגים את היכולות של NIFI, בדוגמה זו הבקשה מתחילה תהליך של קבלת קובץ מאחסון הקבצים המקומי ), לאחר מכן אנו שולחים חזרה תשובה שהבקשה התקבלה, במקביל לתהליך קבלת קובץ מ-FH ולאחר מכן תהליך העברתו דרך FTP ל-FH. כדאי להבהיר שתהליכים מקיימים אינטראקציה זה עם זה באמצעות מה שנקרא flowFile. זוהי ישות הבסיס ב-NIFI המאחסנת תכונות ותוכן. תוכן הוא הנתונים שמיוצגים על ידי קובץ הזרם. כלומר, באופן גס, אם תקבלו קובץ מריבוע אחד ותעבירו אותו למשנהו, התוכן יהיה הקובץ שלכם.

כפי שאתה יכול לראות, תמונה זו מציגה את התהליך הכללי. HandleHttpRequest - מקבל בקשות, ReplaceText - יוצר גוף תגובה, HandleHttpResponse - שולח תגובה. FetchFile - מקבל קובץ מאחסון קבצים, מעביר אותו לריבוע PutSftp - שם קובץ זה על FTP, בכתובת שצוינה. עכשיו עוד על התהליך הזה.

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

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

לאחר מכן, בואו נסתכל על פרמטרי התצורה של ReplaceText של הריבוע. כדאי לשים לב ל-ReplacementValue - זה מה שיוחזר למשתמש בצורה של תגובה. בהגדרות ניתן להתאים את רמת הרישום, ניתן לראות את הלוגים {היכן פרקת את nifi}/nifi-1.9.2/logs, ישנם גם פרמטרים של כשל/הצלחה - על סמך פרמטרים אלו ניתן לווסת את התהליך בכללותו . כלומר, במקרה של עיבוד טקסט מוצלח, תהליך שליחת התגובה למשתמש ייקרא, ובמקרה אחר פשוט נרשום את התהליך הלא מוצלח.

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

סידרנו את הבקשה והתגובה - נעבור לקבלת הקובץ והצבתו על שרת ה-FTP. FetchFile - מקבל קובץ בנתיב שצוין בהגדרות ומעביר אותו לתהליך הבא.

ואז ריבוע PutSftp - מציב את הקובץ באחסון הקבצים. אנו יכולים לראות את פרמטרי התצורה להלן.

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

דוגמה מורכבת יותר

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

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

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

ישנן שתי אפשרויות לטעינת הסקריפט בריבוע זה. הראשון הוא הורדת קובץ עם סקריפט. השני הוא על ידי הוספת סקריפט לתוך scriptBody. עד כמה שידוע לי, ריבוע ה-executeScript תומך במספר שפות - אחת מהן היא גרובי. אני אאכזב מפתחי Java - אי אפשר לכתוב סקריפטים בג'אווה בריבועים כאלה. למי שבאמת רוצה, צריך ליצור ריבוע מותאם אישית משלכם ולהוסיף אותו למערכת NIFI. כל המבצע הזה מלווה בריקוד די ארוך עם טמבורין, בו לא נעסוק במאמר זה. בחרתי בשפה הגרובי. להלן סקריפט בדיקה שפשוט מעדכן בהדרגה את המזהה בהודעת SOAP. חשוב לציין. אתה לוקח את הקובץ מ-flowFile ומעדכן אותו, אל תשכח שאתה צריך להחזיר אותו לשם, מעודכן. כמו כן, ראוי לציין שלא כל הספריות כלולות. יכול לקרות שעדיין תצטרך לייבא את אחת ה-libs. חיסרון נוסף הוא שהתסריט בריבוע הזה די קשה לניפוי באגים. יש דרך להתחבר ל-NIFI JVM ולהתחיל את תהליך איתור הבאגים. באופן אישי, פתחתי אפליקציה מקומית ודימתי קבלת קובץ מהסשן. עשיתי גם ניפוי באגים מקומי. שגיאות שמופיעות בעת טעינת סקריפט די קלות לגוגל ונכתבות על ידי NIFI עצמה ללוג.

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

למעשה, כאן מסתיימת ההתאמה האישית של הריבוע. לאחר מכן, הקובץ המעודכן מועבר לריבוע, אשר אחראי על שליחת הקובץ לשרת. להלן ההגדרות עבור הריבוע הזה.

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

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

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

מקור: www.habr.com