NiFiのAPIを叩いてプログラマブルにフローを制御する

[Apache NiFi](https://nifi.apache.org/)(以下NiFi、読み:ないふぁい)は、[Apache Kafka](https://kafka.apache.org/)、[Apache Storm](http://storm.apache.org/)と共に、データの収集、ストリーミング分析用途のオープンソースをまとめた[Hortonworks DataFlow](http://hortonworks.com/products/data-center/hdf/)(HDF)というパッケージに含まれています。

NiFiにはRDBMSやAWS, Azureなどのクラウド、Hadoopエコシステムなど、多様なデータストアに対応したProcessorという部品が用意されていて、プログラミングなしでシステム間のデータ連携が可能です。

## サンプルデータフロー「郵便番号のナウ」

Advent Calendar向けにいくつか記事を書く予定があるので、サンプルのデータフローを作りました。このフローの詳細については、別の日に紹介しますね。
郵便番号検索APIの[ZipCloud](http://zipcloud.ibsnet.co.jp/doc/api)と、天気情報APIの[OpenWeatherMap](http://openweathermap.org/)を使って、郵便番号に関連する情報を収集してJSONファイルに保存するデータフローです。いわゆる情報の「エンリッチ」ってやつです。

※OpenWeatherMapについては[無料天気予報APIのOpenWeatherMapを使ってみる](http://qiita.com/nownabe/items/aeac1ce0977be963a740)を参考にさせていただきました。

こちらがNiFiのデータフローです。ProcessGroupにまとめて、「調べる」というInputPortを用意し、親のフローから呼べるようにしてあります。

<img width="1244" alt="zipcode-now.png" src="https://qiita-image-store.s3.amazonaws.com/0/20515/f87c3825-f684-15d3-c40c-01edada20e54.png">

## 本日のお題、NiFiのAPIを活用する

さて、NiFiのWeb UIは非常にリッチで、いろんなことができるんですが、UIの操作が必要ということは自動化できないのでは？という質問を良く伺います。

質問に対する答えは、「できます！NiFiのREST APIを使います！」です。

ブラウザからNiFiのWeb UIで実行する操作はすべて、NiFiのREST APIを呼び出しています。これをスクリプトで実行することで、自動化が可能なのです。

## 一発屋フローを実装する

NiFiの自動化が求められる理由の一つに、既存のETLが行うようなJob単位での実行をしたいというのがあります。NiFiはフローをずっと起動しておき、ストリーミング形式でイベントドリブンな処理を行うため、ETLの様なJobスケジューラに慣れ親しんだ方は多少戸惑うようです。

「どこかからデータを取得して、ちょっと加工して、このデータベースに保存して終わる」という一発屋フローは、NiFiではフローの操作なしではできないのです。

「どこかから新しいデータを取得して、ちょっと加工して、このデータベースに保存する」というフローがずーっと稼働しているのがNiFiです。

具体的な例を示しましょう。

前述の「郵便番号のナウ」をテストしたいとしましょう。入力のFlowFileを一回だけ渡して、その結果を確認したいです。その場合、GenerateFlowFileを使ってテスト用のFlowFileを生成することができます:

<img width="330" alt="zipcode-now-test.png" src="https://qiita-image-store.s3.amazonaws.com/0/20515/ef33e81a-12a0-a92f-cbbf-e8cd1a88e217.png">

このGenerateFlowFileをスタートするのですが、問題はNiFiのスケジューラです:

![image](https://qiita-image-store.s3.amazonaws.com/0/20515/d842ded8-58a7-0e25-74ae-cbdeaf84f0a1.png)

NiFiでは、Scheduling StrategyをTimer drivenにしておくと、Run Scheduleの設定間隔でスケジューリングされます。デフォルトでは0 secとなっており、スレッドに空きがあれば常に動くことになります。

今回の目的では、一度だけ実行したいので、1dとし、次に実行されるのは一日後としました。この状態でスタートすれば、テスト用のFlowFileを一個だけ生成し、次の日まで何もしないことができます。

つまり、そのままにしておくと、明日またテストが動いてしまうのです。また、二回目以降のテストを同日に実行したいなら、一度GenerateFlowFileプロセッサをStopして、またStartして、という操作が必要になります。

「1回だけ動くJob的な処理? それってNiFiっぽくないっすね」
というイメージが伝わったら幸いです。

## NiFiのREST APIで解決

それでも、いろいろな都合で一発屋を実装したり、自動化したい場合があるでしょう。その場合はREST APIを使いましょう。

NiFi Web UIをブラウザで操作しながら、ブラウザの開発者ツールを使うと、どの操作をするとどんなHTTPリクエストが飛んでいるのかを観察することができます。

「テスト開始」のGenerateFlowFileを右クリックしてStartすると、`http://localhost:8080/nifi-api/processors/e0f759d1-0158-1000-3ec4-997043625a4c`にPUTリクエストを送っています:

![image](https://qiita-image-store.s3.amazonaws.com/0/20515/638f317a-2586-4127-7344-e39c03876756.png)

ブラウザから送信されたリクエストのペイロードを確認すると、`state: "RUNNING"`がJSONの中に含まれています。これでプロセッサの起動/停止状態を操作しているわけですね。

ちなみに、REST APIのエンドポイント一覧は[NiFiのREST APIドキュメント](https://nifi.apache.org/docs/nifi-docs/rest-api/index.html)に記載されています。

これをcURLコマンドとかで叩いても何とかなりますが、もう少し楽に実行できると良いですね。

## JavaScript向けのAPIクライアントを作ってみました

私もNiFiのテスト自動化など、REST APIを使うことが多いので、Node.jsで自動化スクリプトを書いて実行するための[APIクライアント](https://github.com/ijokarumawak/nifi-api-client-js)をつくってみました。

使えるFunctionの一覧は[このあたりのコード](https://github.com/ijokarumawak/nifi-api-client-js/blob/master/nifi-api-client.js#L64)を見てください。

使い方:

```bash
# Githubからプロジェクトをクローン
git clone https://github.com/ijokarumawak/nifi-api-client-js.git

# インストール
$ cd nifi-api-client-js
$ npm install
```

/tmpにプロジェクトをクローンしたとして、これを使うJSファイルを作成します:

```
/tmp/test-zipcode-now.js
/tmp/nifi-api-client-js (クローンしたAPIクライアント)
```

`test-zipcode-now.js`の内容はこんな感じにします:

```Javascript
// クライアントライブラリを読み込む
var nifiApiClient = require('./nifi-api-client-js');

// 操作するNiFiの接続先設定
var conf = {
  host: 'localhost',
  port: 8080
};

// クライアントを初期化
var nifiApi = new nifiApiClient.NiFiApi(conf);

// 引数でstart/stopを指定
var running = (process.argv.length > 2 && 'start' === process.argv[2]);

// ProcessorのUUIDを指定して操作
nifiApi.updateProcessorState('e0f759d1-0158-1000-3ec4-997043625a4c', running, (err) => {
  if (err) return console.log('Failed.', err);
  console.log('Processor is now', running ? 'running' : 'stopped');
});
```

そして、以下のコマンドで、Processorの起動停止ができました！

```bash
# スタート!
$ node test-zipcode-now.js start
Processor is now running

# ストップ!
$ node test-zipcode-now.js stop
Processor is now stopped
```

NiFiのUIを確認すると起動できたのが確認できます。ちなみに、プロセッサのUUIDは対象をクリックすると、Operateパレットに表示されます:

![image](https://qiita-image-store.s3.amazonaws.com/0/20515/430bf488-43cf-d280-5069-7c46541ecc2d.png)

## まとめ

いかがでしたでしょうか。NiFiのREST APIを使うとどんなことができるのか、、UIでできることはすべてできますね！この仕組を使って、[NiFiのシステムテストを実施する](https://github.com/ijokarumawak/nifi-integration-tests)プロジェクトも個人的に作っています。参考になれば幸いです。