Apache NIFI - مروری کوتاه بر ویژگی ها در عمل

معرفی

این اتفاق افتاد که در محل کار فعلی من مجبور شدم با این فناوری آشنا شوم. با کمی پیش زمینه شروع می کنم. در جلسه بعدی به تیم ما گفته شد که باید یکپارچگی را ایجاد کنیم سیستم شناخته شده. منظور از ادغام این بود که این سیستم معروف درخواست‌ها را از طریق HTTP به یک نقطه پایانی خاص برای ما ارسال می‌کرد و ما، به طرز عجیبی، پاسخ‌ها را در قالب یک پیام SOAP ارسال می‌کردیم. همه چیز ساده و پیش پا افتاده به نظر می رسد. از اینجا نتیجه می گیرد که شما نیاز به ...

کار

3 سرویس ایجاد کنید. اولین مورد از آنها سرویس به روز رسانی پایگاه داده است. این سرویس هنگامی که داده های جدیدی از یک سیستم شخص ثالث می رسد، داده های موجود در پایگاه داده را به روز می کند و یک فایل با فرمت CSV تولید می کند تا آن را به سیستم بعدی منتقل کند. نقطه پایانی سرویس دوم نامیده می شود - سرویس حمل و نقل FTP، که فایل منتقل شده را دریافت می کند، آن را تأیید می کند و از طریق FTP در ذخیره سازی فایل قرار می دهد. سرویس سوم، سرویس انتقال داده مصرف کننده، به صورت ناهمزمان با دو مورد اول کار می کند. درخواستی از یک سیستم خارجی شخص ثالث برای دریافت فایل مورد بحث در بالا دریافت می کند، فایل پاسخ آماده را می گیرد، آن را اصلاح می کند (شناسه، توضیحات، فیلدهای linkToFile را به روز می کند) و پاسخ را در قالب یک پیام SOAP ارسال می کند. یعنی تصویر کلی به شرح زیر است: دو سرویس اول فقط زمانی کار خود را شروع می کنند که داده های به روز رسانی رسیده است. سرویس سوم به طور مداوم کار می کند زیرا مصرف کنندگان اطلاعات زیادی وجود دارد، حدود 1000 درخواست داده در دقیقه. خدمات به طور مداوم در دسترس هستند و نمونه های آنها در محیط های مختلف مانند تست، دمو، پیش تولید و تولید قرار دارند. در زیر نموداری از نحوه عملکرد این خدمات ارائه شده است. اجازه دهید فوراً روشن کنم که برخی از جزئیات برای جلوگیری از پیچیدگی های غیر ضروری ساده شده اند.

تعمیق فنی

هنگام برنامه ریزی برای حل مشکل، ابتدا تصمیم گرفتیم با استفاده از فریم ورک Spring، متعادل کننده Nginx، پایگاه داده Postgres و سایر موارد فنی و نه چندان فنی، برنامه هایی را در جاوا ایجاد کنیم. از آنجایی که زمان توسعه یک راه حل فنی به ما اجازه داد تا روش های دیگری را برای حل این مشکل در نظر بگیریم، نگاه ما به فناوری Apache NIFI افتاد که در محافل خاص مد است. فوراً می گویم که این فناوری به ما امکان می دهد به این 3 سرویس توجه کنیم. این مقاله توسعه یک سرویس انتقال فایل و یک سرویس انتقال داده به مصرف کننده را شرح می دهد، اما اگر مقاله مفید باشد، در مورد سرویس به روز رسانی داده ها در پایگاه داده خواهم نوشت.

این چیست؟

NIFI یک معماری توزیع‌شده برای بارگذاری و پردازش موازی سریع داده‌ها، تعداد زیادی افزونه برای منابع و تبدیل‌ها، نسخه‌سازی تنظیمات و موارد دیگر است. یک امتیاز خوب این است که استفاده از آن بسیار آسان است. فرآیندهای بی اهمیت مانند getFile، sendHttpRequest و موارد دیگر را می توان به صورت مربع نشان داد. هر مربع نشان دهنده فرآیندی است که تعامل آن در شکل زیر قابل مشاهده است. مستندات دقیق تر در مورد تعاملات راه اندازی فرآیند نوشته شده است اینجا ، برای کسانی که روسی صحبت می کنند - اینجا. این مستندات به خوبی نحوه باز کردن و اجرای NIFI و همچنین نحوه ایجاد فرآیندهایی را که به عنوان مربع نیز شناخته می شوند، توضیح می دهد.
ایده نوشتن مقاله پس از یک جستجوی طولانی و ساختار دادن اطلاعات دریافتی به چیزی آگاهانه و همچنین تمایل به آسان کردن زندگی برای توسعه دهندگان آینده متولد شد.

مثال

نمونه ای از نحوه تعامل مربع ها با یکدیگر در نظر گرفته شده است. طرح کلی بسیار ساده است: ما یک درخواست HTTP دریافت می کنیم (در تئوری، با یک فایل در بدنه درخواست. برای نشان دادن قابلیت های NIFI، در این مثال درخواست فرآیند دریافت یک فایل از محل ذخیره فایل محلی را شروع می کند. ، سپس پاسخی مبنی بر دریافت درخواست ارسال می کنیم، به موازات فرآیند دریافت فایل از FH و سپس فرآیند انتقال آن از طریق FTP به FH. شایان ذکر است که فرآیندها از طریق به اصطلاح flowFile با یکدیگر تعامل دارند. این موجودیت پایه در NIFI است که ویژگی ها و محتوا را ذخیره می کند. محتوا داده ای است که توسط فایل جریانی نمایش داده می شود. یعنی به طور کلی اگر فایلی را از یک مربع دریافت کنید و به مربع دیگر منتقل کنید، محتوا فایل شما خواهد بود.

همانطور که می بینید، این تصویر روند کلی را نشان می دهد. HandleHttpRequest - درخواست ها را می پذیرد، ReplaceText - یک بدنه پاسخ تولید می کند، HandleHttpResponse - پاسخی را ارسال می کند. FetchFile - یک فایل را از یک ذخیره سازی فایل دریافت می کند، آن را به مربع PutSftp منتقل می کند - این فایل را در FTP در آدرس مشخص شده قرار می دهد. حالا بیشتر در مورد این فرآیند.

در این صورت درخواست آغاز همه چیز است. بیایید به پارامترهای پیکربندی آن نگاه کنیم.

همه چیز در اینجا کاملاً بی اهمیت است به استثنای StandardHttpContextMap - این نوعی سرویس است که به شما امکان ارسال و دریافت درخواست را می دهد. با جزئیات بیشتر و حتی با مثال، می توانید ببینید - اینجا

در مرحله بعد، اجازه دهید به پارامترهای پیکربندی ReplaceText مربع نگاه کنیم. ارزش توجه به ReplacementValue را دارد - این همان چیزی است که در قالب یک پاسخ به کاربر بازگردانده می شود. در تنظیمات می‌توانید سطح گزارش را تنظیم کنید، می‌توانید گزارش‌ها را ببینید {where you unpacked nifi}/nifi-1.9.2/logs، همچنین پارامترهای شکست/موفقیت وجود دارد - بر اساس این پارامترها می‌توانید روند را به طور کلی تنظیم کنید. . یعنی در صورت موفقیت آمیز بودن پردازش متن، فرآیند ارسال پاسخ به کاربر فراخوانی می شود و در حالتی دیگر به سادگی فرآیند ناموفق را Log می کنیم.

هیچ چیز جالبی در ویژگی های HandleHttpResponse وجود ندارد به جز وضعیت زمانی که یک پاسخ با موفقیت ایجاد می شود.

ما درخواست و پاسخ را مرتب کردیم - اجازه دهید به دریافت فایل و قرار دادن آن در سرور FTP ادامه دهیم. FetchFile - یک فایل را در مسیر مشخص شده در تنظیمات دریافت می کند و آن را به فرآیند بعدی ارسال می کند.

و سپس مربع PutSftp - فایل را در ذخیره سازی فایل قرار می دهد. ما می توانیم پارامترهای پیکربندی را در زیر مشاهده کنیم.

شایان توجه است که هر مربع فرآیند جداگانه ای است که باید راه اندازی شود. ما به ساده ترین مثالی نگاه کردیم که نیازی به سفارشی سازی پیچیده ندارد. در مرحله بعد، ما به روند کمی پیچیده تر نگاه می کنیم، جایی که کمی روی شیارها می نویسیم.

مثال پیچیده تر

به دلیل فرآیند تغییر پیام SOAP، سرویس انتقال داده به مصرف کننده کمی پیچیده تر بود. روند کلی در شکل زیر نشان داده شده است.

در اینجا این ایده نیز چندان پیچیده نیست: ما درخواستی از مصرف کننده دریافت کردیم مبنی بر اینکه به داده نیاز دارد، پاسخی ارسال کردیم که پیامی دریافت کرده است، فرآیند دریافت فایل پاسخ را شروع کردیم، سپس آن را با منطق خاصی ویرایش کردیم و سپس آن را ویرایش کردیم. فایل را در قالب یک پیام SOAP به سرور به مصرف کننده منتقل کرد.

من فکر می کنم نیازی به توصیف مجدد آن مربع هایی که در بالا دیدیم نیست - بیایید مستقیماً به مربع های جدید برویم. اگر نیاز به ویرایش هر فایلی دارید و مربع های نوع ReplaceText معمولی مناسب نیستند، باید اسکریپت خود را بنویسید. این کار را می توان با استفاده از مربع ExecuteGroogyScript انجام داد. تنظیمات آن در زیر ارائه شده است.

دو گزینه برای بارگذاری اسکریپت در این مربع وجود دارد. اولین مورد با دانلود یک فایل با یک اسکریپت است. مورد دوم با قرار دادن یک اسکریپت در scriptBody است. تا آنجا که من می دانم، مربع executeScript از چندین زبان پشتیبانی می کند - یکی از آنها groovy است. من توسعه دهندگان جاوا را ناامید خواهم کرد - شما نمی توانید اسکریپت هایی را در جاوا در چنین مربع هایی بنویسید. برای کسانی که واقعاً می خواهند، باید مربع سفارشی خود را ایجاد کرده و آن را به سیستم NIFI اضافه کنید. کل این عملیات با رقصی کاملا طولانی با تنبور همراه است که در این مقاله به آن نمی پردازیم. من زبان groovy را انتخاب کردم. در زیر یک اسکریپت آزمایشی وجود دارد که به سادگی شناسه را در یک پیام SOAP به‌روزرسانی می‌کند. نکته حائز اهمیت است. شما فایل را از flowFile می گیرید و آن را به روز می کنید، فراموش نکنید که باید آن را در آنجا به روز کنید. همچنین شایان ذکر است که همه کتابخانه ها شامل نمی شوند. ممکن است این اتفاق بیفتد که شما هنوز مجبور باشید یکی از lib ها را وارد کنید. نکته منفی دیگر این است که اشکال زدایی اسکریپت در این مربع بسیار دشوار است. راهی برای اتصال به NIFI JVM و شروع فرآیند رفع اشکال وجود دارد. من شخصاً یک برنامه محلی راه اندازی کردم و دریافت فایل از جلسه را شبیه سازی کردم. اشکال زدایی هم به صورت محلی انجام دادم. خطاهایی که هنگام بارگیری یک اسکریپت ظاهر می شوند برای Google بسیار آسان هستند و توسط خود 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