はじめに

「手を動かして学ぶ！KafkaとPythonによるAWSストリーミングデータ分析入門【30日チャレンジ】」7日目: PythonでConsumerを実装してデータを受信しよう！です。昨日はPythonを使って、高性能なKafkaプロデューサーを実装する方法を学びました。

今日は、その逆の役割を担うコンシューマーの実装に挑戦します。Kafkaからリアルタイムにデータを読み取り、加工・分析する基盤の第一歩です。

1. Pythonコンシューマーの実装

Kafkaコンシューマーは、特定のトピックを購読し、そこからデータを取得します。今回は、昨日のプロデューサーが送信したデータをリアルタイムで受信するスクリプトを作成しましょう。

consumer.pyというファイルを作成し、以下のコードを記述してください。

from confluent_kafka import Consumer, KafkaError, KafkaException
import json
import sys

# Kafkaコンシューマーの設定
conf = {
    'bootstrap.servers': '127.0.0.1:9092',
    'group.id': 'my-python-consumer',
    'auto.offset.reset': 'earliest'
}

# Consumerインスタンスの作成
consumer = Consumer(conf)

# 購読するトピック
topic = "test-topic"

try:
    # 指定したトピックを購読開始
    consumer.subscribe([topic])

    # 無限ループでメッセージを受信
    while True:
        # メッセージを受信（タイムアウト1.0秒）
        msg = consumer.poll(1.0)

        if msg is None:
            # 受信するデータがなかった場合
            continue
        if msg.error():
            # エラーが発生した場合
            if msg.error().code() == KafkaError.PARTITION_EOF:
                # 終端に達したが、これは正常な状態
                sys.stderr.write(
                    f"%% {msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}\n")
            elif msg.error():
                raise KafkaException(msg.error())
        else:
            # 正常にメッセージを受信した場合
            # メッセージのペイロードをデコード
            try:
                message_value = msg.value().decode('utf-8')
                data = json.loads(message_value)
                print(f"Received message: {data}")
            except Exception as e:
                print(f"Failed to decode or parse JSON: {e}")
                print(f"Raw message: {msg.value()}")

except KeyboardInterrupt:
    sys.stderr.write('%% Aborted by user\n')

finally:
    # Consumerをクローズ
    consumer.close()

2. コードの解説と実行

(1) コードの解説

'group.id': 'my-python-consumer': コンシューマーグループIDを指定します。同じグループIDを持つコンシューマーは、トピックのパーティションを共有して並列にデータを処理します。例えば、3つのパーティションがあるトピックを3つのコンシューマーで購読する場合、各コンシューマーが1つずつのパーティションを担当し、処理が分散されます。
'auto.offset.reset': 'earliest': どこからデータの読み取りを開始するかを指定します。
- 'earliest'：トピックの先頭（最も古いオフセット）から読み取ります。
- 'latest'：最新のデータから読み取ります。
consumer.subscribe([topic]): 購読したいトピック名をリストで指定します。
consumer.poll(1.0): 新しいメッセージをポーリング（一定時間待機して取得）します。引数の1.0はタイムアウト時間（秒）です。
エラーハンドリング: msg.error()で、受信中に発生したエラーを確認します。KafkaError.PARTITION_EOFは、トピックの終端に達したことを示す正常なエラーです。
メッセージのデコード: 受信したデータはバイト列なので、.decode('utf-8')で文字列に戻し、json.loads()でPythonの辞書型に変換しています。

(2) 実行方法

DockerでKafkaが起動していることを確認:
docker-compose up -dでKafkaコンテナが起動しているか確認してください。
producer_async.pyを実行:
producer_async.pyを別のターミナルで実行して、Kafkaにデータを送信しておきましょう。
```
python producer_async.py
```
consumer.pyを実行:
新しいターミナルでconsumer.pyを実行します。
```
python consumer.py
```

実行すると、プロデューサーが送信したデータがリアルタイムに表示されるはずです。

コンシューマーはCtrl + Cで停止するまで、メッセージを受信し続けます。

まとめと次回予告

今日は、PythonでKafkaコンシューマーを実装し、リアルタイムにデータを受信する方法を学びました。これで、PythonからKafkaへのデータの送受信という、ストリーミングデータ分析の最も基本的なパイプラインが完成しました。

来週からは、いよいよAWS環境への移行と実践に入ります。

8日目: AWSでストリーミング！Amazon MSKの概要とメリット

お楽しみに！