Theme
今回は前回に引き続き、以下を参考に進めていきます。
https://github.com/ijokarumawak/hdf-tutorials-ja/wiki/Apache-NiFi-Dataflow-Automation-Concepts
前回のを見てない場合は、こちらからご確認いただければと思います。
Part1
今回はKafkaについて見ていきます。
Kafkaとは
Kafkaは大量のメッセージを高速に扱うことができる分散メッセージシステムで、メッセージングキューとして必要な様々な機能があります。
大まかに分けて、Producer(書き込み)、Topic(Queue)、Consumer(読み込み)で構成されます。
ProducerがTopicに書き込み、ConsumerはTopicにデータをとりに行きます。
Reporting ProcessGroupの作成
今回は、前回までのHTTP APIとは別にReportingというProcess Groupを作成します。
参考記事からそのまま言葉をお借りしますが、
単一のグループで作業を進め、大きなデータフローになってくると、管理が煩雑になってしまいます。 チュートリアルでは、以下の２つに分けています:
- データを外部から収集し、共通のフォーマットに変換する部分
- 共通のフォーマットのデータを入力として、Kafkaにメッセージ登録を行う部分
こうすることで、HTTP以外の、TCPやMQTTなどでメッセージを受信するルートを増やす際に、変換部分のみを実装すれば良くなります。
Processの配置
Input Portの追加
他のProcessGroupからFlowFileを渡すために、ReportingグループへInput Portを追加していきます。
(追加後)
PublishKafka_2_0の追加
PublishKafkaに関しては、自分の環境(kafkaのversion)にあったものを設定してください。
私の環境は、ambari2.7.3で入れたものになるので、2.0となっています。
前回同様、ConfigureからPROPATIESに行き、以下を設定します。
-
Kafka Brokers
localhost:6667 -
Topic Name
input -
Delivery Guarantee
Guarantee Replicated Delivery -
Kafka Key
${message.key}
Kafkaメッセージはkeyとvalueを持っています。 Keyには、渡ってきたFlowFileのmessage.key Attributeを利用します。
Valueには、FlowFileのcontentが渡されます。
kafka BrokersのPortについては、ambariからkafkaをインストールした場合は、
Kafka > Configsから確認できます。
SETTINGSタブでは、Automatically terminate relationshipsにおいてsuccessにチェックを入れます。
APPLYを選択した後、LogAttributeも設置し、すべてをConnectionを作成しましょう。
HTTP APIからReportingへデータを流し込む
ルートProcessGroupへと戻り、HTTP APIからReportingへとRelationをつなぎます。
こうすることで、ProcessGroup間はInput PortとOutput Portでデータの連携が可能になります。
Console Consumerで確認しながらテスト
サーバにSSHでログインし、以下のコマンドでConsole Consumerを起動します。
cd /usr/hdp/current/kafka-broker
./bin/kafka-console-consumer.sh --topic input --bootstrap-server localhost:6667 --new-consumer
別terminalで以下を実施します。
# curl -i -X POST -H "Content-type: application/json" -d '{"name": "C", "age": 20}' localhost:9095
#### Postした側 ####
HTTP/1.1 202 Accepted
Date: Tue, 15 Oct 2019 13:59:47 GMT
Transfer-Encoding: chunked
Server: Jetty(9.4.11.v20180605)
{"Result": "succeeded"}
#### consumer ####
20
Ctrl+Cを押すと
Processed a total of 1 messages
と表示されると思います。
これで、Postしたデータに変更を加えて、Kafkaに送信し、そのメッセージを受け取るまでのフローをNifiから操作できました。
次回
次は、このメッセージをStormに送ってリアルタイム分析をしていきます。