Apache NiFiセカンド・ステップのバックアップ(No.6) - .NET 開発基盤部会 Wiki

[ トップ ] [ 新規 | 一覧 | 単語検索 | 最終更新 | ヘルプ ]

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

戻る
- Apache NiFi
- Apache NiFiファースト・ステップ

目次 †

目次
概要
チュートリアル
WebAPIを作成してみる。
参考
- Apache NiFiファースト・ステップ

概要 †

セカンド・ステップでは、

Learning the Ropes of Apache NiFi? · ijokarumawak/hdf-tutorials-ja Wiki
https://github.com/ijokarumawak/hdf-tutorials-ja/wiki/Learning-the-Ropes-of-Apache-NiFi

WebAPIを作成してみる。
HandleHttpRequest? / HandleHttpResponse?のProcessorを使用して、WebAPIを作成してみる。

などをやってみる。

チュートリアル †

出だしのZIPがリンク切れなので、以下から取得。

https://community.hortonworks.com/questions/107442/file-analyze-traffic-pattern-with-apache-nifiasset.html

本ページに添付しておく。

16302-trafficlocs-data-for-simulator.zip

チュートリアル 1 †

「シンプルなNiFi?データフロー構築」をやってみる。

概要
- 車両位置XMLシミュレータからのデータを取り込む、
- フローファイルから車両位置詳細属性を抽出する、
- それらの詳細属性が空でない場合JSONファイルに変換。

リンク切れ
- Lab1-NiFi?-Learn-Ropes.xmlテンプレート
- trafficLocs_data_for_simulator.zip

Processorを追加する。 †

GetFile?
処理するファイルを入力。

UnpackContent?
交通シミュレータZipファイルからフローファイルのコンテンツを解凍

ControlRate?
プロセッサに流すフローファイルのフローのレートを制御。

EvaluateXPath
車両位置データの最終更新タイムスタンプ属性を抽出

SplitXML
親の子要素を複数のフローファイルに分割

UpdateAttribute?
各フローファイルの名称属性を更新

EvaluateXPath
各フローファイルの車両要素から属性（車両ID、方向、緯度、軽度、速度）を抽出

RouteOnAttribute?
各属性がフィルタ条件に一致する場合のみ、フローファイルを後続のフローに流す。

AttributesToJSON
フローファイルから属性を抽出し、JSON形式に変換

MergeContent?
JSONフローファイルのグループをマージ

PutFile?
処理したファイルを出力。

Connectionで接続する。 †

図がリンク切れだけど、コンテキストから推測するに多分、こう。

手順1-1

※ 全てのConnectionのRelationshipsを選択し、ProcessorのTerminateは設定しなかった。

Processorを設定する。 †

「1.1 学習の目的: データフロー作成プロセスの概要」を参考に設定。
（詳細は「Step 2」以降に書いてあるので、必要に応じてソレを参考にする）

GetFile?
- Input Directory : ./data-in

UnpackContent?
- Packaging Format : zip

ControlRate?
- Rate Control Criteria : flowfile count
- Maximum Rate : 3
- setTime Duration : 10 second

EvaluateXPath
- Destination : flowfile-attribute ※ XPath式の結果をフローファイルの属性に保存する
- XPath式
  - Last_Time : //body/lastTime/@time ※ [+]ボタンでプロパティ追加する。

SplitXML : 既定値

UpdateAttribute?
- filename : ${UUID()} ※ [+]ボタンでプロパティ追加する。

EvaluateXPath
- Destination : flowfile-attribute ※ XPath式の結果をフローファイルの属性に保存する
- XPath式
  - Direction_of_Travel : //vehicle/@dirTag ※ [+]ボタンでプロパティ追加する。
  - Latitude : //vehicle/@lat ※ [+]ボタンでプロパティ追加する。
  - Longitude : //vehicle/@lon ※ [+]ボタンでプロパティ追加する。
  - Vehicle_ID : //vehicle/@id ※ [+]ボタンでプロパティ追加する。
  - Vehicle_Speed : //vehicle/@speedKmHr? ※ [+]ボタンでプロパティ追加する。

RouteOnAttribute?
- Terminate Relationships : unmatched ※ ココだけは必要そうなので。ConnectionのRelationshipsも設定。
- Filter_Attributes : 以下の式を記入 ※ [+]ボタンでプロパティ追加する。
```
${Direction_of_Travel:isEmpty():not():and(${Last_Time:isEmpty():not()}):and(${Latitude:isEmpty():not()}):and(${Longitude:isEmpty():not()}):and(${Vehicle_ID:isEmpty():not()}):and(${Vehicle_Speed:equals('0'):not()})}
```

AttributesToJSON
- Destination : flowfile-content
- Attributes List : Vehicle_ID, Direction_of_Travel, Latitude, Longitude, Vehicle_Speed, Last_Time

MergeContent?
- Minimum Number of Entries : 10
- Maximum Number of Entries : 15
- Delimiter Strategy : Text
- Header : [
- Footer : ]
- Demarcator : , {now-press-shift-enter}

PutFile?
- Directory : ./data-out
- Terminate Relationships : failure, success

ココまでの設定が適切であれば、Processorのwarningマークはすべて取れているハズ。

手順1-1

※ RouteOnAttribute?だけ、Terminate Relationships : unmatched を設定した。

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

入力ファイル・フォルダsにZIPを放り込むと、上手いこと、JSONが出力される。
しかし、ファイル名がUUIDのものが出力される。これはマージ後のフローファイルもフローに流れているためと解る。
MergeContent?のTerminate Relationships : original を設定してテストして上手く動作することを確認する。

チュートリアル 2 †

「地理情報でデータフローをエンリッチメント」をやってみる。

概要
- Google Places Nearby APIをNiFi?から利用し、
- 車両の移動に応じて周辺情報を表示。

リンク切れ
- Lab2-NiFi?-Learn-Ropes.xmlテンプレート

チュートリアル 3 †

「NextBus?ライブストリームの取込」をやってみる。

概要
- Google Places Nearby APIをNiFi?から利用し、
- 車両の移動に応じて周辺情報を表示。

リンク切れ
- Lab2-NiFi?-Learn-Ropes.xmlテンプレート

WebAPIを作成してみる。 †

HandleHttpRequest? / HandleHttpResponse?のProcessorを使用して、WebAPIを作成してみる。
HandleHttpRequest? / HandleHttpResponse?の利用方法についてはコレが参考になる。

参考 †

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