手を動かして学ぶ！KafkaとPythonによるAWSストリーミングデータ分析入門【30日チャレンジ】 - 6日目: PythonでシンプルなProducerを実装してみよう

はじめに

「手を動かして学ぶ！KafkaとPythonによるAWSストリーミングデータ分析入門【30日チャレンジ】」6日目: PythonでシンプルなProducerを実装してみよう！です。昨日はconfluent-kafka-pythonライブラリをセットアップし、簡単なプロデューサーのスクリプトでKafkaにデータを送信できることを確認しました。

今日は、さらに一歩進んで、より実用的なプロデューサーの実装に挑戦します。エラーハンドリングや非同期送信といった、本番環境で役立つテクニックを学びましょう。

---

1. プロデューサーの実装：非同期送信とコールバック

昨日のスクリプトでは、producer.flush()で同期的にデータを送信していましたが、これでは大量のデータを扱う際にパフォーマンスが低下する可能性があります。Kafkaのクライアントは非同期送信をサポートしており、これにより高いスループットを実現できます。

以下のコードをproducer_async.pyとして作成してください。

from confluent_kafka import Producer
import json
import time
import random

# Kafkaブローカーの設定
conf = {'bootstrap.servers': '127.0.0.1:9092'}

# Producerインスタンスの作成
producer = Producer(conf)

# 配信レポート（Delivery Report）コールバック関数
def delivery_report(err, msg):
    """
    メッセージがKafkaに正常に送信されたか、エラーが発生したかを通知するコールバック関数。
    """
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

# データを連続的に生成してKafkaに送信
topic = "test-topic"
for i in range(10):
    user_id = random.randint(1000, 9999)
    event_type = random.choice(['click', 'view', 'purchase'])
    timestamp = int(time.time())

    data = {
        'user_id': user_id,
        'event_type': event_type,
        'timestamp': timestamp,
        'value': i
    }
    json_data = json.dumps(data)
    
    try:
        # 非同期送信。コールバック関数を指定。
        producer.produce(
            topic,
            key=str(user_id).encode('utf-8'),
            value=json_data.encode('utf-8'),
            callback=delivery_report
        )
        print(f"Attempting to send message {i}...")

    except Exception as e:
        print(f"Error producing message: {e}")

    # 一定間隔でメッセージを送信
    time.sleep(1)

# 送信中のすべてのメッセージが完了するまで待機
producer.flush()
print("All messages flushed and delivery reports received.")

---

2. コードの解説と実行

(1) コードの解説

• delivery_report関数: この関数は、producer.produce()のコールバック関数として渡されます。メッセージが正常に送信された場合やエラーが発生した場合に、Kafkaクライアントによって自動的に呼び出されます。これにより、メッセージごとに送信結果を確認できます。
• データの生成: ループ内でuser_idやevent_typeといったダミーデータを生成しています。これにより、実際のストリーミングデータに近い形でテストができます。
• producer.produce(): メッセージを送信する主要なメソッドです。
  • key=str(user_id).encode('utf-8')：今回はuser_idをキーとして指定しています。これにより、同じユーザーIDのデータは常に同じパーティションに送信されます。データの順序性を保ちたい場合に非常に重要です。
  • callback=delivery_report：メッセージの送信結果をdelivery_report関数で受け取るように設定します。producer.produce()自体はすぐに返ってくるため、ノンブロッキング（非同期）で動作します。
• producer.flush(): producer.produce()が非同期であるため、プログラムの終了前にflush()を呼び出す必要があります。これにより、バッファ内に残っているすべてのメッセージがKafkaに送信されるまで処理をブロックします。

(2) 実行方法

DockerでKafkaが起動していることを確認し、以下のコマンドでスクリプトを実行します。

python producer_async.py

実行すると、メッセージが1秒ごとに送信され、そのたびにdelivery_report関数が呼び出されて送信結果が表示されます。

データの確認

4日目に使ったコンソールコンシューマーを別のターミナルで起動しておくと、リアルタイムでデータが流れてくるのが確認できます。

docker exec -it kafka /bin/bash
kafka-console-consumer.sh --topic test-topic --bootstrap-server 127.0.0.1:9092 --from-beginning

---

まとめと次回予告

今日は、PythonでKafkaプロデューサーを実装する際のベストプラクティスである非同期送信とコールバック関数の使い方を学びました。これにより、高いパフォーマンスと信頼性を両立させることができます。

明日は、これとは逆の操作、つまりPythonでKafkaからデータを受信する方法を学んでいきます。

7日目: PythonでConsumerを実装してデータを受信しよう

お楽しみに！