Nifi + Kafka + Strom を用いたデータ処理ハンズオン(1) ~Nifiの基礎部分~ 

# Theme
今回、以下を参考に新しいバージョンでのNifi, Kafka, Stormをつかった検証を行っていきたいと思います。
https://github.com/ijokarumawak/hdf-tutorials-ja/wiki/Apache-NiFi-Dataflow-Automation-Concepts

主に、今回はNifiのパートになります。

# Apache NiFiとは
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/15cf4020-6655-881a-c450-e32e6a291849.png)


システム間のデータフローを自動化し管理するためのOSS

# Nifi 基本操作

## login
NifiはNifiを動かしているサーバの`9090`portを使用します。

http://127.0.0.1:9090

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/ba9ef537-ab29-30ff-21e0-cd6f491fd0c2.png)

## Process Groupの作成

赤く囲った部分をドラッグして、方眼上に持っていき、リリースしてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/3d5d82f8-2954-70e2-b236-b8f42d71c039.png)

今回、名前をHTTP APIとします。

ADDを押し、少し時間を置くと以下のようにProcess Groupが生成されます。
そうしたら、ダブルクリックをして中に入ってください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/9fabc363-50fb-50f6-d3a5-e20086684b0d.png)


## 処理を定義

HTTP APIに入ると、左下の`Operator`が`HTTP API`に変わります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/8f10e7e5-bab1-a9fb-e7e0-e9f78d3043aa.png)

### HandleHttpRequestプロセッサの追加
検索場所にて`Handle`と打つと、`HandleHttpRequest`というのがあるので、それをダブルクリックします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/424d9845-7dbd-76c3-15d6-23b3db08ebd6.png)

(追加後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/1bcd39a6-6959-7c8e-664e-e6a2c0556783.png)


上の様に表示されましたら、追加したHandleHttpRequestプロセッサを右クリックし、`Configure`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/edc6db7d-c9e6-d70e-0393-a1038a49abd7.png)

`Property`タブに移動し、`Listening Port`を`9095`にします。
その後、APPLYを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/45bdee55-777d-ad81-5bbe-a27c31ff276a.png)

### ReplaceTextの追加
こちらでは、HTTPレスポンスで返すボディの文字列を設定します。
前と同様に、Processorを追加します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/71eb3148-d1ca-0ba0-5fd3-0b3b7150d0aa.png)

追加されたProcessorに対して、右クリックし、`Configure`を選択します。
`Property`タブに移動し、`Replacement Value`を`{"Result": "Succeeded"}`,
`Replacement Strategy`を`Always Replace`
に変更し、APPLYを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/9f811425-30d1-c2f9-f716-3978eff4c985.png)

### HandleHttpResponseプロセッサの追加

次は、HandleHttpResponceプロセッサを追加します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/7eaf30aa-7e0d-f8e8-5599-ab811018fbed.png)

(追加後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/5df42427-47f9-8916-63f2-422ea9dbd8ce.png)

そうしたら、HandleHttpResponseのstatusコードを設定していきます。
HandleHttpResponce上で右クリックをし、`Configure`を選択します。
その後、`PROPERTIES`にて`HTTP Status Code`を決め打ちで`202(Accepted)`
にします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/895ebc72-37b5-f164-1d0d-62e6a611af32.png)


### 3つのプロセッサの接続

NiFiのデータフローは、プロセッサ同士を`Relationship`でつなぐことで流れを生成します。 
プロセッサの中心からマウスをドラッグし、接続先のプロセッサでドロップすることで、つなぐことができます。

`From Processor`と`To Processor`を確認し、`For relationships`は`Success`を選択し、ADDを選択します。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/20ff801d-fd2c-2bf0-063a-34d5931a22c4.png)

(接続後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/4aafeac5-fbe1-09a6-74d2-2a36b284e851.png)


### LogAttributeにFailureを流す

プロセッサの処理が失敗した場合、渡ってきたFlowFileは`Relationship`の`failure`に流されます。
失敗内容が確認できるように、LogAttributeプロセッサを追加します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/db79cbd8-2628-37af-4371-2628176a7b54.png)

(追加後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/9509f461-6416-1d35-89bd-870bb67fb493.png)

その後、'Relationship'で接続します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/6bd4ab3c-c6db-5dcb-29d8-47b54ef4d55a.png)


### 不要なRelationshipのAuto-Terminate

NiFiの内部を流れるFlowFileは、行き先がなくなる (終点までたどり着く) と、削除されます。
今回の設定では、`HandleHttpResponse`と`LogAttribute`において、処理がSuccessになった後の処理がありません。
なので、そのあとはNifiFlowを自動で消去する設定を入れてあげます。

以下の`Relationship`に関して、プロセッサで右クリックをし、`Configure`を選択します。'SETTINGS`タブから`Automatically Terminate Relationships`において、`success`にチェックを入れます。

- HandleHttpResponse: success
- LogAttribute: success

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/fabc8795-5758-fe33-dd3f-8fcc89000bd2.png)

### Http Context Mapの作成
HandleHttpRequest上で右クリックをし、`Configure`を選択します。
その後`PROPERTIES`タブの`Http Context Map`のプルダウンより、`Create new service...`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/0f1b1c29-10a3-ac3e-153e-806d5e397039.png)

すると、次の画面のようになります。
`CREATE`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/d5ae7fe3-907d-cc11-d54b-6e67b7724b83.png)

その後、APPLYを選択すると、以下の画面になります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/cfe5c7ba-de50-59a7-2b01-e6d7a04f26f8.png)

右側、真ん中の雷のようなアイコンを選択すると以下の画面が表示されるので、`ENABLE`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/926491dd-5bfc-a6eb-c79d-9dc041827079.png)

同様にHandleHttpRequestのHTTP Context Mapでも同じControllerServiceを指定します。 
これでプロセッサに必要な設定が完了します。

### フローの開始
左側にある`Operate`の再生ボタンを選択し、フローを開始します。
一度、方眼上でクリックをし、特定のプロセッサから選択を外してから行ってください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/68ad65ce-43fa-3e4f-2b20-ae93e5161662.png)

全て起動が成功すると、以下の様になります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/6635a853-8708-6587-acb9-f584b07da22c.png)


### cURLで簡単なテスト

```
curl -i -X POST -H "Content-type: application/json" -d '{"name": "Nifi", "process": "test"}' localhost:9095
HTTP/1.1 202 Accepted
Date: Mon, 14 Oct 2019 10:47:51 GMT
Transfer-Encoding: chunked
Server: Jetty(9.4.11.v20180605)

{"Result": "succeeded"}
```

NifiでもIn/Outから、処理の流れが確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/4e80060a-a878-b6d7-a0a0-afebfcb2ad20.png)

# データの変換

## プロセッサの作成
POSTで受信したデータをKafkaに登録する前に整形していきます。

次のプロセッサを追加します:

- EvaluateJsonPath: FlowFileのcontentをJSONとしてパースし、JSONPathで任意の要素の値を抽出します。
抽出結果をFlowFileのAttributeに保存します。

- ReplaceText: Kafkaに登録する際、FlowFileのContentがメッセージのvalueとなります。
  抽出したJSON内の値を再びFlowFileのContentに戻します。

### EvaluateJsonPath
Processの追加をします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/465c1e80-32ef-f676-f7ef-8d651fc547c7.png)

(追加後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/9fb2bb64-4790-4066-5e61-82e241713600.png)

その後、右クリックで`Configure`を選択します。
`PROPERTIES`にて以下の設定をしていきます。

- Destination
flowfile-attribute
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/fa50e2ce-ad57-4036-8ec2-301a64dad819.png)

以下2つは、右上の`+ボタン`を押して追加していきます。
- message.key
$.name
- message.value
$.age

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/aff0304c-4f3b-dc70-0df6-f6fb0b374abd.png)

これによって、NifiFlowの中に`message.key: $.name(つまりJsonでのnameに対するvalue)`というkey-value型のデータが格納されます。
`message.value`も同様です。

### ReplaceText
Processの追加をします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/054cc1d8-8e9f-b2b2-1ffc-41d521c3d860.png)

(追加後)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/ac36a24b-15b2-027e-ddad-15cd9ca16dfa.png)

その後、右クリックで`Configure`を選択します。
`PROPERTIES`にて以下の設定をしていきます。

- Replacement Value
${message.value}
これは、$で始まっていますが、[Apache NiFi Expression Language Guide](https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html)というもので、JsonPathではありません。

- Replacement Strategy
Always Replace
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/0977ba3c-709b-0f99-02c6-6846ceef9ec4.png)


## Output Portの追加

以下の箇所から、Output Portを作成し、名前を入力して`ADD`を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/8feb1919-88fc-c8af-184d-1e410430cdf1.png)

(追加後)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/5651be5e-f88b-3430-06f0-5419a873c068.png)

## processの接続
今回作成したprocessを接続して、前回したテストを実行してみます。

EvaluateJsonPathからReplaceTextへのRelationshipでは、`matched`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/0e04dbd0-f69f-ed09-6a3e-7cb3ab937d6d.png)

最終的に以下のようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/983f7b2e-ef95-8bd0-4563-6b7e331209fd.png)

では、起動してテストをしてみます。
テスト結果

```
curl -i -X POST -H "Content-type: application/json" -d '{"name": "Nifi", "process": "test"}' localhost:9095
HTTP/1.1 202 Accepted
Date: Mon, 14 Oct 2019 13:21:56 GMT
Transfer-Encoding: chunked
Server: Jetty(9.4.11.v20180605)

{"Result": "succeeded"}
```

先ほどと同様の結果です。
では、今回作ったフローの流れはどうなっているかというと
フローのIn/Outはそれぞれ1になっています。

Output Portの前のConnectionの上で右クリックして`List Queue`を選択してみます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/0e20e833-9569-3b62-3d0a-995c54b55a9a.png)

すると、下の様にQueueの中に先ほどのNifiFlowが格納されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/416540/f71e5ab0-8871-3fb2-7388-9fc03a1e0f8a.png)

# 次回
次回では、このNifiからKafkaに対してメッセージを送りたいと思います。