Apache NIFI - 実際の機会の概要

導入

たまたま、現在の職場でこのテクノロジーに触れる必要がありました。 少し背景から始めます。 次の会議で、私たちのチームは、との統合を構築する必要があると言われました。 既知のシステム。 統合とは、このよく知られたシステムが HTTP 経由でリクエストを特定のエンドポイントに送信し、奇妙なことに、SOAP メッセージの形式で応答を送り返すことを意味していました。 すべてが単純で些細なことのように思えます。 このことから、次のことが必要になります...

タスク

3つのサービスを作成します。 1000 つ目はデータベース更新サービスです。 このサービスは、サードパーティのシステムから新しいデータが到着すると、データベース内のデータを更新し、CSV 形式のファイルを生成して次のシステムに転送します。 XNUMX 番目のサービスのエンドポイントは FTP トランスポート サービスと呼ばれ、転送されたファイルを受信し、検証して、FTP 経由でファイル ストレージに置きます。 XNUMX 番目のサービスである消費者データ転送サービスは、最初の XNUMX つのサービスと非同期で動作します。 サードパーティの外部システムから上記のファイルを受信するリクエストを受信し、準備ができた応答ファイルを取得して変更し (ID、説明、linkToFile フィールドを更新)、SOAP メッセージの形式で応答を送信します。 つまり、全体像は次のようになります。最初の XNUMX つのサービスは、更新用のデータが到着したときにのみ作業を開始します。 XNUMX 番目のサービスは、情報の利用者が多く、XNUMX 分あたり約 XNUMX 件のデータ要求があるため、常に動作します。 サービスは常に利用可能であり、そのインスタンスはテスト、デモ、運用前、本番などのさまざまな環境に配置されています。 以下は、これらのサービスがどのように機能するかを示す図です。 不必要な複雑さを避けるために、いくつかの詳細が簡略化されていることをすぐに明確にさせてください。

技術の深化

この問題の解決策を計画する際、私たちはまず Spring フレームワーク、Nginx バランサー、Postgres データベース、その他の技術的およびそれほど技術的でないものを使用して Java でアプリケーションを作成することにしました。 技術的なソリューションを開発する時間ができたため、この問題を解決するための他のアプローチを検討することができたので、私たちは特定のサークルで流行している Apache NIFI テクノロジーに注目しました。 このテクノロジーにより、これら 3 つのサービスを認識できるようになったとすぐに言います。 この記事では、コンシューマー向けのファイル転送サービスとデータ転送サービスの開発について説明しますが、この記事が役立つ場合は、データベース内のデータを更新するサービスについても書きます。

此れは何ですか

NIFI は、データの高速な並列読み込みと処理、ソースと変換用の多数のプラグイン、構成のバージョン管理などを行うための分散アーキテクチャです。 素晴らしい利点は、非常に使いやすいことです。 getFile、sendHttpRequest などの単純なプロセスは正方形として表すことができます。 各四角形はプロセスを表しており、その相互作用を以下の図に示します。 プロセス設定の対話に関するより詳細なドキュメントが作成されました ここで 、ロシア語を話す人のために - ここで。 このドキュメントには、NIFI を解凍して実行する方法と、スクエアとも呼ばれるプロセスを作成する方法が完全に説明されています。
記事を書くというアイデアは、長い検索と受け取った情報を意識的なものに構造化すること、そして将来の開発者の作業を少し楽にしたいという願望の後に生まれました。

例

正方形がどのように相互作用するかの例を考えてみましょう。 一般的なスキームは非常に単純です: HTTP リクエストを受信します (理論的には、リクエストの本文にファイルが含まれます。NIFI の機能を示すために、この例では、リクエストはローカル ファイル ストレージからファイルを受信するプロセスを開始します) ）、FHからファイルを受信する処理と、FTP経由でFHにファイルを移動する処理を並行して、リクエストを受信した旨の応答を返します。 プロセスはいわゆる flowFile を通じて相互に対話することを明確にする価値があります。 これは、属性とコンテンツを保存する NIFI の基本エンティティです。 コンテンツは、ストリーム ファイルによって表されるデータです。 つまり、大まかに言うと、あるスクエアからファイルを受信し、それを別のスクエアに転送すると、そのコンテンツがあなたのファイルになります。

ご覧のとおり、この図は一般的なプロセスを示しています。 HandleHttpRequest - リクエストを受け入れ、ReplaceText - 応答本文を生成し、HandleHttpResponse - 応答を送信します。 FetchFile - ファイル ストレージからファイルを受信し、それを正方形に転送します。 PutSftp - このファイルを FTP 上の指定されたアドレスに置きます。 このプロセスについて詳しく説明します。

この場合、リクエストがすべての始まりです。 その構成パラメータを見てみましょう。

ここでの内容は、StandardHttpContextMap を除いてすべて非常に簡単です。StandardHttpContextMap は、リクエストの送受信を可能にするサービスの一種です。 さらに詳しく、例も含めて説明すると、次のようになります。 ここで

次に、正方形の ReplaceText 構成パラメーターを見てみましょう。 ReplacementValue に注意する価値があります。これは、応答の形式でユーザーに返されるものです。 設定ではログのレベルを調整でき、ログ {nifi を解凍した場所}/nifi-1.9.2/logs を確認できます。また、失敗/成功パラメータもあります。これらのパラメータに基づいて、プロセス全体を調整できます。 。 つまり、テキスト処理が成功した場合は、ユーザーに応答を送信するプロセスが呼び出されますが、別の場合は、失敗したプロセスを単にログに記録します。

HandleHttpResponse プロパティには、応答が正常に作成されたときのステータスを除いて、特に興味深いものはありません。

リクエストとレスポンスを整理しました。次に、ファイルの受信と FTP サーバーへの配置に進みます。 FetchFile - 設定で指定されたパスでファイルを受信し、次のプロセスに渡します。

次に、PutSftp 四角形 - ファイルをファイル ストレージに配置します。 以下に設定パラメータを示します。

各四角形は、起動する必要がある別個のプロセスであるという事実に注意を払う価値があります。 複雑なカスタマイズを必要としない最も単純な例を見ていきました。 次に、もう少し複雑なプロセスを見て、溝に少し書き込みます。

より複雑な例

コンシューマへのデータ転送サービスは、SOAP メッセージを変更するプロセスにより、少し複雑になることが判明しました。 一般的なプロセスを次の図に示します。

ここでも、考え方は特に複雑ではありません。消費者からデータが必要であるというリクエストを受信し、メッセージを受信したという応答を送信し、応答ファイルを受信するプロセスを開始し、それを特定のロジックで編集して、サーバーへの SOAP メッセージの形式でファイルをコンシューマに転送しました。

上で見た四角形について再度説明する必要はないと思います。新しい四角形に直接進みましょう。 ファイルを編集する必要があり、通常の ReplaceText タイプの四角形が適していない場合は、独自のスクリプトを作成する必要があります。 これは、ExecuteGroogyScript スクエアを使用して実行できます。 その設定を以下に示します。

この四角形にスクリプトをロードするには XNUMX つのオプションがあります。 XNUMX つ目は、スクリプトを含むファイルをダウンロードすることです。 XNUMX つ目は、scriptBody にスクリプトを挿入する方法です。 私の知る限り、executeScript 広場はいくつかの言語をサポートしています。そのうちの XNUMX つは groovy です。 Java 開発者はがっかりするでしょう - このような四角形では Java でスクリプトを書くことはできません。 本当にそうしたい場合は、独自のカスタム スクエアを作成して NIFI システムに追加する必要があります。 この操作全体には、タンバリンを使ったかなり長いダンスが伴いますが、この記事では扱いません。 私はグルーヴィーな言語を選びました。 以下は、SOAP メッセージ内の ID を単純に増分更新するテスト スクリプトです。 注意することが重要です。 flowFile からファイルを取得して更新します。更新してそこに戻す必要があることを忘れないでください。 すべてのライブラリが含まれているわけではないことにも注意してください。 いずれかのライブラリをインポートする必要がある場合があります。 もう XNUMX つの欠点は、この広場のスクリプトのデバッグが非常に難しいことです。 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 プロセスを使用するためのいくつかのオプションを検討しました。 それらはどのように相互作用し、その本当の利点は何でしょうか? 考慮されている例はテスト用であり、実際の戦闘で起こることとは若干異なります。 この記事が開発者にとって少しでも役立つことを願っています。 ご清聴ありがとうございました。 ご質問がございましたら、お書きください。 答えてみます。

出所： habr.com