はじめに
コンソールプロデューサーで、イベントとキーの仕組みを確認できました。アプリケーションには、もう1つ区別すべきことがあります。送信 API を呼び出すと、ブローカーが確認する前に、レコードがクライアントのローカルキューに入る場合があります。その時点で終了するプログラムは、レコードを配信していないのに成功を報告してしまうかもしれません。
用意されたローカルの Apache Kafka ブローカーと、インストール済みの confluent-kafka Python クライアントを使います。このパッケージはクライアントライブラリであり、Confluent Cloud アカウントは不要です。辞書を JSON/UTF-8 のバイト列にシリアライズし、注文キーを送信して、配信コールバックと最終的なキューの状態を調べます。到達できないローカルアドレスを使って、時間制限付きの失敗テストも行います。
この独立した演習には、3つのパーティションを持つトピックと2つの入力用サンプルが用意されています。コース後半のアクティビティは、Python の関数、ループ、辞書、JSON、スクリプトの実行を前提とします。これらの準備には Python 学習パスを利用できます。Kafka のインストールとサービス設定は準備済みです。Terminal を使用してください。
辞書を送信用のキーと値に変換する
このステップでは、Kafka に接続する前に変換を確認します。6件のイベントを含む入力から始めます。
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
各行は完全な JSON オブジェクトです。json.loads はそのテキストを Python の辞書に変換します。json.dumps は辞書を JSON テキストに変換し、encode("utf-8") はテキストをクライアント API に適したバイト列に変換します。キーは order_id から別途エンコードします。event_id ではありません。
次の小さなプレビューを保存して実行します。引用符付きのヒアドキュメントは、Python ソースをそのまま保存します。
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
辞書、JSON テキスト、バイト列の表現を探してください。Python の bytes の表示は b で始まりますが、この表示用の記法は送信する値の一部ではありません。最初のキーは order-4001 をエンコードしたものになるはずです。JSON は名前付きのイベントフィールドを保持し、UTF-8 は文字のエンコーディングを提供します。
まだイベントは送っていません。次のステップでは、このバイト列をプロデューサーに渡し、ブローカーが確認した時点を調べます。
送信して実際の配信をすべて確認する
このステップでは、引数処理と配信コールバックを備えた完全なプロデューサーを実行します。Producer でクライアントを設定し、bootstrap.servers で最初の接続先を指定します。acks=all は、現在同期しているレプリカからの確認応答を要求します。この単一ブローカー環境にはレプリカが1つあるため、この設定だけで複数ブローカーの耐障害性を実証できるわけではありません。指定された murmur2_random の振り分け設定は、パーティション数を固定したまま、以前の Java コンソールの例と互換性のあるキーの振り分けを維持します。
プログラムの重要な違いを読み取ってください。produce はレコードをローカルのキューに追加します。poll はコールバックを処理し、flush はキューに入った処理の完了を待ちながらコールバックも処理します。
コールバックは、配信結果が利用可能になったときに呼び出してもらうため、クライアントライブラリに渡す関数です。callback=delivery は関数自体を渡し、その場で呼び出すわけではありません。poll(0) は待機せずに利用可能なコールバックを処理し、flush は未完了の送信も待ちます。コールバックは、ブローカーで確認されたトピック・パーティション・オフセット、または失敗を記録します。flush の後、プログラムは残ったレコード、失敗、確認件数を調べてから成功を宣言します。失敗コールバックは収集され、プロセスが0以外の結果で終了する原因になります。表示だけして無視することはありません。
用意された argparse 部分は、コマンドラインオプションを args の属性に対応付けます。たとえば --input は args.input、--delivery-timeout-ms は args.delivery_timeout_ms になります。Path(args.input).open(...) は選択したローカルファイルを開き、ループは空ではない各行から JSON イベントを1件ずつ読みます。このクライアントを使うために、自分で引数パーサーを設計する必要はありません。
完全なプログラムを保存します。パーサーとコールバックの構造は用意されており、シリアライズと produce 呼び出しで、直前に確認したバイト列を使用します。
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
プロデューサーを実行し、両方の出力ストリームを保存します。> は標準出力をログに送り、2>&1 は標準エラー出力も同じ保存先に送るため、コールバックの失敗や接続の診断情報も残せます。
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
6行の DELIVERED コールバックと Delivered records: 6 があるはずです。各コールバックの実際のパーティションとオフセットは変わる場合があります。produce の戻り値が成功しただけでは、これらの確認応答が得られたとは言えません。
すでに学んだコンソールコンシューマーで、ブローカーを独立して確認します。
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
py1 から py6 を探し、キーがそれぞれの order_id 値と一致することを確認します。独立した読み取りによって、アプリケーションの配信レポートを実際のブローカーレコードと結び付けられます。固定のパーティション番号や、表示全体の順序は指定しません。
新しいイベントの1フィールドを変えて同じプロデューサーを使う
このステップでは、同じアプリケーションで新しいイベントを1件送ります。用意された次のイベントのサンプルには、新しい event_id の py7 が付いているため、まだ送信されていません。channel フィールドを確認します。
cat /home/labex/project/data/next-event.jsonl
まだ送っていないイベントの channel だけを online から store に変え、別の作業用ファイルに保存します。このイベントも order-4001 を参照し、sequence 3 で以前のイベントに続きます。
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
新しい成功コールバックが1件あり、独立した読み取りでは保持された7件が見えるはずです。py7 の channel は store で、py1 から py6 の内容は元のままであるはずです。関連する order-4001 のイベントは、同じ振り分け先のパーティション内で順番に並びます。パーティションをまたぐ表示順は変わる場合があります。
プロデューサーの実装は変えていません。固定の学習用の答えを埋め込む代わりに、入力パスを指定することで、新しいデータをシリアライズして配信できました。
到達できないアドレスで時間制限付きの失敗を観察する
このステップでは、キューへの追加を配信と取り違えていないことを確かめます。同じクライアントは、localhost:1 に Kafka ブローカーがなくてもローカルキューを作成できます。用意された配信タイムアウトのオプションでメッセージ完了までを3000ミリ秒に制限し、flush はさらに短い一定の余裕を加えます。実際のブローカーは引き続き localhost:9092 にあります。
意図的に間違えたアドレスへ、1件のイベントを含むファイルを送ります。次のコマンドは失敗することを想定しています。
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
別のコマンドで上書きされる前に、直ちに終了状態を保存します。$? は直前のコマンドの終了状態で、0は成功、0以外は失敗です。printf はそれをテキストの1行として保存します。
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
0以外の状態、接続の診断情報、DELIVERY_FAILED コールバック、そして成功の最終件数ではなく Delivery incomplete が見えるはずです。メッセージが失敗してキューが空になった場合にも、プログラムはコールバックの失敗を確認しています。
失敗したアドレスへの送信で、実際のトピックに2件目の py7 は追加されないはずです。以前に成功した7件はすべて利用できます。接続先を修正して業務イベントを再送するかどうかの判断と、キューに入ったレコードが確認されたと考えることは別です。この演習の意図的な失敗では、実際のブローカーには何も送っていません。次は Python コンシューマーで受信した JSON をデコードして保存し、読み取り位置をコミットします。
まとめ
辞書と注文キーを UTF-8 のバイト列にシリアライズし、実際の Python クライアントで送信して、成功コールバックと実際のブローカー上の位置が一致することを確認しました。新しいイベントの内容を変えても同じプログラムを使え、到達できないアドレスへの送信は、誤って成功とされず時間制限内に0以外の状態で失敗しました。次は Python コンシューマーで、デコード、ファイル出力、書き込み後のコミットという動作を明示します。
