みなさん、こんにちは！

タスクは次のとおりです。上の図に示すフローがあり、これを N 台のサーバーにロールアウトする必要があります。 アパッチNiFi。 フロー テスト - ファイルが生成され、別の NiFi インスタンスに送信されます。 データ転送は、NiFi サイト間プロトコルを使用して行われます。

NiFi サイト間 (S2S) は、NiFi インスタンス間でデータを転送する安全で高度にカスタマイズ可能な方法です。 S2S の仕組みを確認する ドキュメンテーション S2S が参照できるように NiFi インスタンスを忘れずに設定することが重要です。 ここで.

S2S を使用したデータ転送に関しては、XNUMX つのインスタンスはクライアントと呼ばれ、XNUMX つ目はサーバーと呼ばれます。 クライアントがデータを送信し、サーバーがそれを受信します。 それらの間のデータ転送を設定するには XNUMX つの方法があります。

1. プッシュ。 データは、リモート プロセス グループ (RPG) を使用してクライアント インスタンスから送信されます。 サーバー インスタンスでは、入力ポートを使用してデータが受信されます。
2. プル。 サーバーは RPG を使用してデータを受信し、クライアントは出力ポートを使用して送信します。

ローリングのフローは Apache レジストリに保存されます。

Apache NiFi Registry は、フロー ストレージおよびバージョン管理ツールを提供する Apache NiFi のサブプロジェクトです。 一種のGIT。 レジストリのインストール、構成、操作に関する情報は、次の場所にあります。 公式ドキュメント。 保存用のフローはプロセスグループにまとめられ、この形式でレジストリに保存されます。 これについては後ほど説明します。

開始時に、N が小さい場合、フローは適切な時間内に手動で配信および更新されます。

しかし、N が大きくなるにつれて、さらに多くの問題が発生します。

1. フローを更新するにはさらに時間がかかります。 すべてのサーバーにアクセスする必要があります
2. テンプレートの更新中にエラーが発生しました。 ここでは更新されましたが、ここでは忘れられていました
3. 同様の操作を多数実行する場合の人的エラー

これらすべてのことから、プロセスを自動化する必要があるという事実がわかります。 この問題を解決するために次の方法を試しました。

1. NiFi の代わりに MiNiFi を使用する
2. NiFi CLI
3. NiPyAPI

MiNiFiの使用

ApacheMiNify は Apache NiFi のサブプロジェクトです。 MiNiFy は、NiFi と同じプロセッサを使用するコンパクトなエージェントで、NiFi と同じフローを作成できます。 エージェントの軽量化は、とりわけ、MiNiFy にはフロー構成用のグラフィカル インターフェイスがないという事実によって実現されます。 MiNiFy にはグラフィカル インターフェイスがないため、minifi でのフロー配信の問題を解決する必要があります。 MiNiFy は IOT で積極的に使用されているため、多くのコンポーネントがあり、最終的な minifi インスタンスにフローを配信するプロセスは自動化する必要があります。 おなじみの作業ですよね？

別のサブプロジェクトである MiNiFi C2 Server は、この問題の解決に役立ちます。 この製品は、展開アーキテクチャの中心となることを目的としています。 環境の構成方法 - で説明されています。 この記事 ハブレに関する情報は問題を解決するのに十分です。 MiNiFi は C2 サーバーと連携して、その構成を自動的に更新します。 このアプローチの唯一の欠点は、C2 サーバー上にテンプレートを作成する必要があることです。レジストリへの単純なコミットだけでは十分ではありません。

上記の記事で説明されているオプションは機能しており、実装は難しくありませんが、次のことを忘れてはなりません。

1. minifi には nifi のプロセッサーがすべて搭載されているわけではありません
2. Minifi の CPU バージョンは、NiFi の CPU バージョンよりも遅れています。

この記事の執筆時点での NiFi の最新バージョンは 1.9.2 です。 最新の MiNiFi バージョンのプロセッサ バージョンは 1.7.0 です。 プロセッサーを MiNiFi に追加できますが、NiFi プロセッサーと MiNiFi プロセッサーのバージョンが異なるため、機能しない可能性があります。

NiFi CLI

で判断する 説明 公式 Web サイトのツール。これは、フロー配信またはプロセス管理の分野で NiFI と NiFi レジストリ間のやり取りを自動化するためのツールです。 このツールをダウンロードして始めてください。 故に.

ユーティリティを実行する

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

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

必要なフローをレジストリからロードするには、バスケットの識別子 (バケット識別子) とフロー自体 (フロー識別子) を知る必要があります。 このデータは、cli または NiFi レジストリ Web インターフェイスを通じて取得できます。 Web インターフェースは次のようになります。

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 インスタンスが他のインスタンスにデータを送信する必要があります。 データをサーバーに転送するためにプッシュ方式が選択されたと仮定します。 データ転送を整理するには、追加したリモート プロセス グループ (RPG) でデータ転送を有効にする (送信を有効にする) 必要があります。これは、フローにすでに含まれています。

CLI やその他のソースのドキュメントでは、データ転送を有効にする方法が見つかりませんでした。 これを行う方法を知っている場合は、コメントに書き込んでください。

バッシュもあるし、最後まで行く準備はできているので、活路を見出します！ NiFi API を使用すると、この問題を解決できます。 次のメソッドを使用して、上記の例から ID を取得します (この場合は 7f522a13-016e-1000-e504-d5b15587f2f3)。 NiFi API メソッドの説明 ここで.

本文では、次の形式の JSON を渡す必要があります。

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

「機能する」ために入力する必要があるパラメータ:
状態 — データ転送ステータス。 データ転送を有効にする場合は TRANSMITTING を使用でき、無効にする場合は STOPPED を使用できます。
バージョン - プロセッサのバージョン

バージョンは作成時にデフォルトで 0 に設定されますが、これらのパラメータは次のメソッドを使用して取得できます。

bash スクリプトのファンにとっては、この方法が適しているように思えるかもしれませんが、私にとっては難しいです。bash スクリプトは私の好みではありません。 私の意見では、次の方法の方が面白くて便利です。

NiPyAPI

NiPyAPI は、NiFi インスタンスと対話するための Python ライブラリです。 ドキュメントページ ライブラリを操作するために必要な情報が含まれています。 クイックスタートについては、 プロジェクト 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

次に、このプロセス グループが既に追加されているかどうかを理解することが重要です。 プロセス グループは座標によって配置されるため、プロセス グループの上に XNUMX 番目のプロセスが重なる場合に状況が発生する可能性があります。 確認したところ、それは可能です 🙂 追加されたすべてのプロセスグループを取得するには、メソッドを使用します

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 ライブラリの開発者に手紙を書いてアドバイス/助けを求めることにしました。 開発者が私に返答し、私たちは問題について話し合ったところ、「何かを確認する」時間が必要だと書いてきました。 そして数日後、私の起動時の問題を解決する Python 関数が書かれた電子メールが届きました。 当時の NiPyAPI のバージョンは 0.13.3 で、当然ながらそのようなものは何もありませんでした。 しかし、ごく最近リリースされたバージョン 0.14.0 では、この機能はすでにライブラリに含まれています。 会う

nipyapi.canvas.set_remote_process_group_transmission

そこで、NiPyAPI ライブラリの助けを借りて、レジストリに接続し、フローをロールアップし、プロセッサとデータ転送も開始しました。 次に、コードを精査し、あらゆる種類のチェックを追加し、ログを記録するだけです。 しかし、それは全く別の話です。

私が検討した自動化オプションの中で、後者が最も効率的であるように思えました。 まず、これは依然として Python コードであり、補助プログラム コードを埋め込んでプログラミング言語の利点をすべて享受できます。 次に、NiPyAPI プロジェクトは積極的に開発されており、問題が発生した場合は開発者に問い合わせることができます。 第三に、NiPyAPI は、複雑な問題を解決する際に NiFi と対話するためのより柔軟なツールです。 たとえば、フロー内のメッセージ キューが現在空かどうか、およびプロセス グループを更新できるかどうかを判断します。

それだけです。 NiFi でのフロー配信を自動化するための 3 つのアプローチと、開発者が遭遇する可能性のある落とし穴について説明し、配信を自動化するための実用的なコードを提供しました。 もしあなたが私と同じようにこのトピックに興味があるなら - ぴょんぴょん！

出所： habr.com