Apache Kafkaを触った


MDC Advent Calendar 2019 の15日目です。

## Kafkaとは何か
LinkedInが作ったOSSの分散メッセージングシステム（メッセージングキュー）。
高スループット、高スケーラビリティ。
Java(Scala)で書かれている。

Producer、Broker、Consumerの3つのコンポーネントで構成される。
ProducerからストリーミングされたデータをConsumerへ中継する。対障害のためにデータの永続化もする。送達保証も実現。
　　・ Producer: メッセージを配信する
　　・ Broker: ProducerからConsumerへメッセージの受け渡しをする
　　・ Consumer: メッセージを受け取る




※仔細な仕組みやアーキテクチャ等については以下が参考になります。
　・[Apache Kafkaの概要とアーキテクチャ](https://qiita.com/sigmalist/items/5a26ab519cbdf1e07af3)
　・[Apache Kafkaに入門した](https://deeeet.com/writing/2015/09/01/apache-kafka/)

## 何に使うのか
ユースケースとしては以下のようなものが挙げられます。

1. システムがサイロ化するのを防ぐためにデータハブとしてアーキテクチャに組み込む（マイクロサービスとかで）
2. Fluentdなどと連携してログ収集に使う
3. Webサイトのユーザのページ移動とかを収集してWebアクティビティ分析に使う
4. IoTデバイスのセンサーの値を集約し、可視化や分析、他のデバイスの制御などに使う
5. ビッグデータ、機械学習、etc

具体的なところだと以下。
・[LINEの大規模データパイプライン](https://logmi.jp/tech/articles/320330)
・[Yahooリアルタイム検索](https://www.slideshare.net/techblogyahoo/kafkastreamplatform)
・[大手ヘルスケアIT企業 Cerner社のKafka活用事例](https://blog.cloudera.co.jp/%E5%A4%A7%E6%89%8B%E3%83%98%E3%83%AB%E3%82%B9%E3%82%B1%E3%82%A2it%E4%BC%81%E6%A5%AD-cerner%E7%A4%BE%E3%81%AEkafka%E6%B4%BB%E7%94%A8%E4%BA%8B%E4%BE%8B-b7a9b2c9e680)

## 動かしてみる
簡単なサンプルを作って試してみます。
今回はKafka本体はKafka-dockerを使って環境構築をして、ProducerとConsumerのクライアント側はkafka-pythonを使いました。

### Kafka-dockerのインストール・起動

[公式](http://wurstmeister.github.io/kafka-docker/)にあるとおり、kafka-dockerをダウンロードして、docker-compose.ymlの``KAFKA_ADVERTISED_HOST_NAME``にdocker hostのIPアドレスを書いたあと、


~~~sh
docker-compose up -d
~~~

すればOKです。

参考： [kafka in dockerのチュートリアル](https://qiita.com/psyashes/items/e50bdfe35a2e7778986d)


### Producerを実装する
ほんとうはTwitter Striming APIみたいなデータをとってきたり、IoTのセンサーの値を取得したかったですが、今回は用意がないので一旦適当に数値を取れるものを、ということでマウスのx座標を取得して1秒おきにKafkaに送付するスクリプトを書きました。

KafkaProducerの引数bootstrap_serverに渡す値は、docker-compose.ymlにも書いたdocker hostのIPアドレスと、kafkaのコンテナに割り当てられたPort番号を指定します。
割り当てられてるPort番号はdocker psで確認できます。
<img width="813" alt="スクリーンショット 2019-12-14 21.42.04.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/269282/4af1fa81-31a6-84a0-8dd7-3c82bfe74012.png">
上の場合は32783がそれです。
以下が作成したProducer側のソースコードです。
``procuer.send(``で``test``というTopicに現在のマウスのx座標を投げています。

~~~python
from kafka import KafkaProducer
import pyautogui
import time

def main():
    producer = KafkaProducer(bootstrap_servers='{Docker HostのIPアドレス}:{Port}')
    while True:
        result = producer.send('test', str(pyautogui.position().x).encode()).get(timeout=60)
        print(result)
        time.sleep(1)

if __name__ == '__main__':
    main()

~~~



### Consumerを実装する
次にConsumer側の実装です。同様にKafkaのIPアドレスとPortを指定します。
``for message in consumer:``でKafkaからデータを逐次pullしてきてくれます。

~~~python
from kafka import KafkaConsumer

def main():
    consumer = KafkaConsumer(
            'test', 
            bootstrap_servers=['{Docker HostのIPアドレス}:{Port}'])

    for message in consumer:
        print("x = " + message.value.decode())

if __name__ == '__main__':
    main()
~~~

### 実行
下の左側でProducer側、右側でConsumerを実行しています。
Producer側からKafkaに送った値(マウスのx座標)をConsumer側でKafkaから取得して表示できているのが確認できます。

![test.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/269282/3646989f-0780-4b57-d1f8-00be051c8758.gif)

## 今後やりたいこと
・ラズパイ、Arduinoとか使ってセンサーの値を組み込む。
・取得したデータをグラフにしたり、解析したりする。