Apache NiFiファースト・ステップ

「.NET 開発基盤部会 Wiki」は、「Open棟梁Project」,「OSSコンソーシアム .NET開発基盤部会」によって運営されています。

• 戻る

目次 †

概要 †

ファーストステップ @ Windows。基本的なことは、コチラで確認して下さい。

インストール †

ダウンロードする。 †

• 以下からダウンロード（Windowsならzipファイル形式のもの）

• Apache NiFi? Downloads
  https://nifi.apache.org/download.html

• 最新の、nifi-toolkitではない方をダウンロード。
  （nifi-toolkitは多分色々なツールが同梱されてる）

• ダウンロードしたらMD5 or SHA1などのハッシュ値をチェックする。
  （HTTPSではないので、バイナリがオカシクなることが有り得る。）

  certutil -hashfile ...nifi-1.6.0-bin.zip MD5
  certutil -hashfile ...nifi-1.6.0-bin.zip SHA1

インストールする。 †

任意のパスに解凍すればイイらしい。とりあえず、Cドラ直下に解凍。
（JMeterっぽい。ZIP64の可能性があるので解凍ツールに注意する。）。

起動する。 †

binフォルダ以下の「run-nifi.bat」バッチを管理者実行。
（これもまたJMeterっぽい。設定ファイルなどは取り合えずスルー）。

• 必要に応じて、Java のランタイム (JRE)のインストールや、パス設定を行う。
• 必要に応じて、confフォルダ以下のnifi.propertiesでポート番号を変更する。
  （私の環境で、Docker for windowsが8080を使用していてUnauthorizedでハマった。
  故に、nifi.propertiesのnifi.web.http.portで、ポート番号を8081に変更。）

最も簡単な利用 †

画面にアクセスする。 †

• 既定値 : http://localhost:8080/nifi/
• 変更後 : http://localhost:8081/nifi/

※ 改めてやってみると、「run-nifi.bat」実行後のロードが長い（2-3分？）。

以下の辺りを理解しつつ、次へ進む。

• Getting Started with Apache NiFi? > I Started NiFi?. Now What?
  https://nifi.apache.org/docs/nifi-docs/html/getting-started.html#i-started-nifi-now-what

Processorを追加する。 †

• Processorの追加により、データフローを作成を開始する。
• Processorについては、コチラを参照。

追加する。 †

• ヘッダのComponentToolBar?のProcessorアイコンを方眼紙風のキャンバスにD&D。
• 追加するProcessorを選択できるダイアログが表示される。
• Processorを選択して[Add]ボタンを押下（必要に応じて[Filter]可能）

• ここでは、ファースト・ステップにお似合いのProcessorを追加する。
  • [GetFile? Processor]
  • [LogAttributes? Processor]

設定する。 †

• Processorをダブルクリック（若しくは、右クリック -> configure）で設定を行う。

• [GetFile? Processor]の設定を、[Property]タブから行う。
  

• ポイント
  • 太字のプロパティは必須のプロパティ
  • 必須プロパティが構成されるまで、Processorは開始できない。

• 設定値
  • Input Directory : ./data-in

• [LogAttributes? Processor]の設定を、[Settings]タブから行う。
  

• ポイント
  末端のProcessorになるので、FlowFile?の処理が完了した場合、
  データを削除することを考慮する必要がある。

• 設定値
  • Automatically Terminate Relationships : checked to [success]
    

使い方を確認する。 †

• 問題がある場合は、warningマークが表示される。
  ここでは、接続が無いため、warningマークが表示される。

• また、Processorを右クリックし、
  [View usage]を選択すると、Processorのヘルプが表示される。

• 次にConnectionで接続をする。

Connectionで接続する。 †

接続する †

Connectionで接続する場合、以下のように操作する。

設定する †

するとダイアログが表示されるので

• [Settings]タブから設定する。

• Connectionに名前を付ける [Name] : hogehoge

• Relationshipの設定 : 既定値

• Dataの有効期限とQueueの設定
  • [FlowFile? Expiration] : 既定値
  • [Back Pressure Object Threshold] : 既定値
  • [Back Pressure Data Size Threshold] : 既定値

• キューの優先順位の設定
  1つ以上のPrioritizerを設定し、
  1つ目のPrioritizerで優先順位が同じだった場合、
  2つ目のPrioritizerで優先順位が再評価される。
  • [Selected Prioritizers] :
    FirstInFirstOutPrioritizer?
    OldestFlowFileFirstPrioritizer?

• [Detail]タブ
  • Relationshipの設定 : 既定値

• ここまでの設定で、Processorは停止状態になる。
  

データフローを実行する。 †

実行 †

• ぞれぞれ、Processorを選択してOperatorの再生ボタンを押下する。
  

• すると、以下の状態になり、データフローを実行できる。
  

• この状態で、data-inフォルダにファイルを放り込むと確実に処理されていることを確認できるのだが、
  LogAttribute? 1.6.0 の仕様が No description provided. のため、最後まで流れたのかどうかを全く確認できない。

修正 †

そこで急遽、PutFile?というProcessorを追加・接続して、以下のように設定して動かしてみる。

• PutFile? Processor
  • Directory : ./data-out
  • [Automatically Terminate Relationships] : checked to [failure] and [success]

• Connection
  • [Name] : hogehoge

確認 †

すると、data-inフォルダに放り込んだファイルがdata-outフォルダに出力されることを確認できる。

※ 後日談：LogAttributeについての仕様を発見した。

分岐、分割・集約辺りの利用 †

分割 †

前述の、GetFile - PutFile の例にSplitText? Processorを追加して、
以下のようなデータフローを定義して分割を評価できる。

データフロー †

※ なお、上記のようにConnectionを曲げたい場合は、矢印上でダブル・クリックする。

ポイント †

• Processor
  • SplitText?
    Line Split Count : 2

• PutFile?
  • PutFile?(Connection 2-1 に接続)
    Directory : ./data-out1
  • PutFile?(Connection 2-2 に接続)
    Directory : ./data-out2

• Connection
  • [Name] : 2-1
    [Relationship] : original, failure
  • [Name] : 2-2
    [Relationship] : splits

動作確認 †

• Processorを選択していない状態でOperatorの再生ボタンを押下する。

• data-inフォルダに放り込んだファイルがdata-out1, 2フォルダに出力されることを確認できる。

• data-out1 : data-inに放り込んだファイル

• data-out2 : SplitText?で抽出した先頭の2行のテキストを含むファイル

分割・集約 †

SplitContent?、MergeContent? Processorを追加、
以下のようなデータフローを定義して分割・集約を評価できる。

データフロー †

ポイント †

• Processor

• GetFile?
  Directory : ./data-in

• PutFile?
  • PutFile?(Connection 3-1 に接続)
    Directory : ./data-temp
  • PutFile?(Connection 3-2 に接続)
    Directory : ./data-out

• 分割・集約

• SplitContent?（[;]区切りで分割）
  Byte Sequence : 3B ※ [;]を意味する。
  Automatically Terminate Relationships : original

• UpdateAttribute?（ファイル名変更）
  filename : ${UUID()} ※ [+]ボタンでプロパティ追加する。

• MergeContent?（統合） : 既定値

• Connection
  • [Name] : 2-1
    [Relationship] : splits
  • [Name] : 2-2
    [Relationship] : splits
  • [Name] : 3-1
    [Relationship] : success
  • [Name] : 3-2
    [Relationship] : failure, merged, original

動作確認 †

• Processorを選択していない状態でOperatorの再生ボタンを押下する。

• data-inフォルダに放り込んだファイルがdata-temp、data-outフォルダに出力されることを確認できる。

• data-temp
  • Byte Sequenceで分割されたファイル
  • ファイル名は、UpdateAttribute?により、UUIDに変更される。

• data-out
  • 分割後、再び集約されたファイル。
  • Byte Sequenceで分割後、マージされるため、Byte Sequenceが削除されている。

変換・処理辺りの利用 †

前述の、GetFile - PutFile の例にCompressContent? Processorを追加して、
以下のようなデータフローを定義して変換・処理を評価できる。

データフロー †

ポイント †

Processor †

• GetFile?
  Directory : ./data-in

• PutFile?
  • PutFile?(Connection 2 に接続)
    Directory : ./data-temp
  • PutFile?(Connection 4 に接続)
    Directory : ./data-out

• 変換・処理

• CompressContent?
  • Mode : compress
  • Compression Format : gzip

• CompressContent?
  • Mode : decompress
  • Compression Format : gzip

Connection †

• [Name] : 1
  [Relationship] : success
• [Name] : 2
  [Relationship] : failure, success
• [Name] : 3
  [Relationship] : failure, success
• [Name] : 4
  [Relationship] : failure, success

動作確認 †

• Processorを選択していない状態でOperatorの再生ボタンを押下する。

• data-inフォルダに放り込んだファイルがdata-temp、data-outフォルダに出力されることを確認できる。

• data-temp
  CompressContent?（compress）で圧縮されたデータ・ファイル

• data-out
  CompressContent?（decompress）で解凍されたデータ・ファイル

その他、基本操作 †

Processor・Connectionの配置 †

ヘッダのComponentToolBar?のProcessor・Connectionアイコンを方眼紙風のキャンバスにD&D。

Connection †

曲げたい場合 †

矢印上でダブル・クリックする。

Queueのクリア †

• Terminateしたフローファイルがキューに詰まることがある。
• この場合、Connectionを右クリックして、Empty Queueする。

Template †

セーブ・ロード †

• セーブ
  • OperatorからCreate Templateする。

• ファイルに出力する場合、
  • 右上メニューからTemplateを選択し、
  • Create TemplateをDownloadする。

• ロード
  • ファイルから入力する場合、OperatorからUpload Templateする。
  • ヘッダのComponentToolBar?のTemplateアイコンを方眼紙風のキャンバスにD&D。

ProcessGroup?に格納 †

テンプレートを保存する前に、以下の手順を踏む。

• ヘッダのComponentToolBar?のProcessGroup?アイコンを方眼紙風のキャンバスにD&D。
• ProcessGroup?に名称を付与し、ProcessGroup?上に他のProcessor＋ConnectionをD&D。

これにより、データフローがProcessGroup?として纏まる。

サーバーの実行 †

・・・

セカンド・ステップ †

参考 †

参考資料 †

Apache NiFi - Processor †