Apache Kafkaを触った #Python

MDC Advent Calendar 2019 の15日目です。

Kafkaとは何か

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

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

※仔細な仕組みやアーキテクチャ等については以下が参考になります。
　・Apache Kafkaの概要とアーキテクチャ
　・Apache Kafkaに入門した

何に使うのか

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

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

具体的なところだと以下。
・LINEの大規模データパイプライン
・Yahooリアルタイム検索
・大手ヘルスケアIT企業 Cerner社のKafka活用事例

動かしてみる

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

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

公式にあるとおり、kafka-dockerをダウンロードして、docker-compose.ymlのKAFKA_ADVERTISED_HOST_NAMEにdocker hostのIPアドレスを書いたあと、

docker-compose up -d

すればOKです。

参考： kafka in dockerのチュートリアル

Producerを実装する

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

KafkaProducerの引数bootstrap_serverに渡す値は、docker-compose.ymlにも書いたdocker hostのIPアドレスと、kafkaのコンテナに割り当てられたPort番号を指定します。
割り当てられてるPort番号はdocker psで確認できます。

上の場合は32783がそれです。
以下が作成したProducer側のソースコードです。
procuer.send(でtestというTopicに現在のマウスのx座標を投げています。

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してきてくれます。

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から取得して表示できているのが確認できます。

今後やりたいこと

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