はじめに
「手を動かして学ぶ！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の概要とメリット
お楽しみに！