Apache NiFiセカンド・ステップのバックアップ(No.10)

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?
- 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()})}
```
- Terminate Relationships : unmatched をチェック
  　※ ココだけは必要そうなので。ConnectionのRelationshipsも設定。

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 : , {press-shift+enter}
  ※ 波括弧内は要するに改行コードを意味している。
  NiFi?のGUIで改行コードを入力するのにshiftが必要ということ。

PutFile?
- Directory : ./data-out/filtered_transitLoc_data
- Terminate Relationships : failure, success をチェック

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

※ RouteOnAttribute?だけ、Terminate Relationships : unmatched をチェックした。

↑

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

Processorを選択していない状態でOperatorの再生ボタンを押下する。
入力ファイル・フォルダにZIPを放り込むと、上手いこと、出力ファイル・フォルダJSONが出力される。
しかし、ファイル名がUUIDのものが出力される。これはマージ後のフローファイルもフローに流れているためと解る。
MergeContent?のTerminate Relationships : original をチェックしてテストして上手く動作することを確認する。

↑

チュートリアル 2 †

「地理情報でデータフローをエンリッチメント」をやってみる。エンリッチメント＝拡充的な意味か。
https://github.com/ijokarumawak/hdf-tutorials-ja/wiki/Ropes-of-Apache-NiFi:-Tutorial-2

概要
- Google Places APIをNiFi?から利用し、
- 車両の移動（車両位置情報XMLのシミュレーションデータ）に応じて周辺情報を表示。

リンク切れ
- Lab2-NiFi?-Learn-Ropes.xmlテンプレート
  リンク切れだったので演習後のものを以下に添付した。
  Lab2-NiFi-Learn-Ropes.xml
  ※ Google Places APIのAPIキーは削除してある。

↑

準備 †

Google Places APIのAPIキーを取得しておく。
https://developers.google.com/places/web-service/get-api-key

↑

Processorを追加する。 †

以下のProcessorを追加する。

InvokeHTTP
Google Places APIから車両ロケーション付近のJSON場所データを取得。

EvaluateJsonPath?
JSON場所データからneighborhoods_nearbyとcityデータ要素を抽出して属性に設定。

RouteOnAttribute?
neighborhoods_nearbyとcity属性が空でないフローファイルをルーティング。

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

MergeContent?
JSON形式のフローファイルのグループを配列としてマージ

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

↑

Connectionで接続する。 †

チュートリアル 1のRouteOnAttribute?から分岐して、
合流せずに、PutFile?で終わるデータフローを作成。

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

↑

Processorを設定する。 †

詳細が「Step 2」以降に書いてあるので、必要に応じてソレを参考にする。

InvokeHTTP

Remote URL : 以下を設定する。

https://maps.googleapis.com/maps/api/place/nearbysearch/json?location=${Latitude},${Longitude}&radius=500&type=neighborhood&key=[取得したAPIキー]

Terminate Relationships : response以外を全てチェック

EvaluateXPath
- Destination : flowfile-attribute ※ JSONPath式の結果をフローファイルの属性に保存する
- Return Type : json
- JSONPath式
  - city : $.results[0].vicinity ※ [+]ボタンでプロパティ追加する。
  - neighborhoods_nearby : $.results[*].name ※ [+]ボタンでプロパティ追加する。

RouteOnAttribute?
- RouteNearbyNeighborhoods? : 以下の式を記入 ※ [+]ボタンでプロパティ追加する。
```
${city:isEmpty():not():and(${neighborhoods_nearby:isEmpty():not()})}
```
- Terminate Relationships : unmatched をチェック
  　※ ココだけは必要そうなので。ConnectionのRelationshipsも設定。

AttributesToJSON
- Destination : flowfile-content
- Attributes List : Vehicle_ID, city, Latitude, Longitude, neighborhoods_nearby, Last_Time

MergeContent?
- Minimum Number of Entries : 10
- Maximum Number of Entries : 15
- Delimiter Strategy : Text
- Header : [
- Footer : ]
- Demarcator : , {press-shift+enter}
  ※ 波括弧内は要するに改行コードを意味している。
  NiFi?のGUIで改行コードを入力するのにshiftが必要ということ。

Terminate Relationships : original をチェック

PutFile?
- Directory : ./data-out/nearby_neighborhoods_search
- Terminate Relationships : failure, success をチェック

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

↑

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

Processorを選択していない状態でOperatorの再生ボタンを押下する。
入力ファイル・フォルダにZIPを放り込むと、上手いこと、出力ファイル・フォルダJSONが出力される。

↑

チュートリアル 3 †

「NextBusライブストリームの取込」をやってみる。
https://github.com/ijokarumawak/hdf-tutorials-ja/wiki/Ropes-of-Apache-NiFi:-Tutorial-3

概要
- 車両位置情報XMLのシミュレーションデータを生成するデータフローセクションを、
- NextBusのライブストリームデータを取込セクションで置換。

リンク切れ
- Lab3-NiFi?-Learn-Ropes.xmlテンプレート
  リンク切れだったので演習後のものを以下に添付した。
  Lab3-NiFi-Learn-Ropes.xml
  ※ Google Places APIのAPIキーは削除してある。

↑

Processorを追加する。 †

以下のProcessorを追加する。

GetHTTP
- GetFile?、UnpackContent?、ControlRate?プロセッサを削除。
- これらをGetHTTPプロセッサで置き換え、入力をシミュレーションデータから、NextBus?ライブストリームに変更。

↑

Connectionで接続する。 †

GetHTTPとEvaluateXPathを接続したデータフローを作成。

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

↑

Processorを設定する。 †

以下のProcessorを設定する。

GetHTTP
- URL : http://webservices.nextbus.com/service/publicXMLFeed?command=vehicleLocations&a=sf-muni&t=0
- Filename : vehicleLoc_SF_OceanView_${now():format("HHmmssSSS")}.xml
- Run Schedule : 1 sec

※ APIの仕様が変わったのか、データが取得できなかったので、パラメタの"r=M"を削除した。

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

↑

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

Processorを選択していない状態でOperatorの再生ボタンを押下する。
1 sec毎に、NextBusのHTTPレスポンスから入力、更出力ファイル・フォルダJSONが出力される。

※ キューが詰まっている部分が赤く表示される。

↑

追加（自習）コンテンツ †

↑

WebAPIを作成してみる。 †

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

↑

Processorを追加する。 †

以下のProcessorを追加する。

HandleHttpRequest?
HTTPクライアントからのHTTPリクエストを受け取る。

ReplaceText?
HTTPリクエストからHTTPレスポンスで返すボディの文字列を設定。

HandleHttpResponse?
HTTPクライアントにHTTPレスポンスを返す。

LogAttribute?
ReplaceText?とHandleHttpResponse?のfailureをこちらに流す。

↑

Connectionで接続する。 †

以下のように接続する。

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

↑

Processorを設定する。 †

以下のProcessorを設定する。

HandleHttpRequest?
- Listening Port : 9095
- HTTP Context Map :
  - DDLから[Create new service...]を選択する。
  - [Add Controller Service]ダイアログで[APPLY]ボタンをクリック。
  - [StandardHttpContextMap?]Controller Serviceが追加され[→]が表示される。
  - コレを押下し、[Enable]アイコンをクリックし[StandardHttpContextMap?]を有効化。

ReplaceText?
- Replacement Value : {"Result": "OK"}
- Replacement Strategy : Always Replace

HandleHttpResponse?
- HTTP Status Code : 202
- HTTP Context Map :
  DDLから、先ほど作成した[StandardHttpContextMap?]を選択する。
- Terminate Relationships : success をチェック

LogAttribute?
- Terminate Relationships : success をチェック

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

※ Controller Serviceとは、コチラ?を参照。
　StandardHttpContextMap?で、Processor間でのHttpContext?を共有する。

↑

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

cURLで以下を実行する。

>curl -i -XPOST -H "Content-type: application/json" -d "{"name": "C", "age": 20}" localhost:9095

以下のように結果が返り、JSONがファイルに保存される。

HTTP/1.1 202 Accepted
Date: Mon, 09 Jul 2018 05:05:32 GMT
Transfer-Encoding: chunked
Server: Jetty(9.4.3.v20170317)

{"Result": "OK"}

↑

参考 †

↑

Apache NiFiセカンド・ステップ のバックアップ(No.10)

目次 †

概要 †

チュートリアル †

チュートリアル 1 †

Processorを追加する。 †

Connectionで接続する。 †

Processorを設定する。 †

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

チュートリアル 2 †

準備 †

Processorを追加する。 †

Connectionで接続する。 †

Processorを設定する。 †

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

チュートリアル 3 †

Processorを追加する。 †

Connectionで接続する。 †

Processorを設定する。 †

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

追加（自習）コンテンツ †

WebAPIを作成してみる。 †

Processorを追加する。 †

Connectionで接続する。 †

Processorを設定する。 †

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

スクレイピング処理をしてみる。 †

Processorを追加する。 †

Connectionで接続する。 †

Processorを設定する。 †

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

参考 †

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

Apache NiFiセカンド・ステップのバックアップ(No.10)