はじめに
ある注文が作成され、別の注文が作成され、その後で最初の注文の支払いが完了したとします。現在の状態を示すテーブルには、各注文の最新の状態しか残らないかもしれません。イベントは発生した出来事を表すため、この3つの事実をすべて履歴として活用できます。新しい注文に反応するシステムには、こうした事実をアプリケーション間で受け渡す仕組みが必要です。
Apache Kafka はイベントレコードを保存して配信します。プロデューサーはレコードを送り、ブローカーは受信して保存するサーバー、トピックはレコードのストリームに付けた名前、コンシューマーはレコードを読み取るクライアントです。Kafka は異なる時間や速度で動作するアプリケーションをつなぎます。イベントからレポートを計算する処理は別のツールでも実行できます。このコースでは、まず確実な配信と読み取り位置に注目します。
トピックを作成し、別々のターミナルでプロデューサーとコンシューマーを観察して、最初のコンシューマーが終了した後に同じイベントを再び読み取ります。用意された単一ブローカー環境ではレプリカを1つ使用します。クラスターの構築は別のテーマです。最初の6つのアクティビティに必要なのは、基本的なターミナル操作だけです。後半の Python クライアントのアクティビティでは、関数、ループ、辞書、JSON、スクリプトの実行を使います。これらは Python 学習パスで準備できます。Java、Hadoop、Hive、Spark、外部アカウントの知識や用意は不要です。Terminal サービスを使用してください。
注文イベントの保存先を作成する
このステップでは、用意された Kafka サービスが応答することを確認し、トピックを作成します。最初のシェルをターミナル A として使います。作業ディレクトリに移動しましょう。
## Terminal A
cd /home/labex/project
コマンドラインクライアントはインストール済みです。--bootstrap-server には、クラスターの情報を取得するために最初に接続するブローカーのアドレスを指定します。ここでの localhost:9092 は用意されたローカルサービスで、外部アカウントではありません。トピックの一覧を表示し、実際にブローカーから応答を得られることを確かめます。
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --list
初期状態の学習環境には、アプリケーション用のトピックがまだない場合があります。orders-first を作成します。トピックは、レコードを順序付きで保存するログであるパーティションを1つ以上含みます。最初は流れを追いやすいようにパーティションを1つにします。--replication-factor 1 は、用意された単一ブローカーに適した指定です。
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-first --partitions 1 --replication-factor 1
新しいトピックの情報を調べます。
## Terminal A
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-first
トピック名、PartitionCount: 1、ReplicationFactor: 1 を探してください。ID、リーダーの識別子、表示形式は環境により異なります。トピックを作成したことで保存先ができました。この時点では注文を送信しておらず、別のサーバーをインストールしたわけでもありません。
短時間だけ動作するプロデューサーやコンシューマーが起動・終了しても、ブローカーは動作を続けます。次は、この独立した動作を利用してイベントの到着を観察します。
一方のクライアントが送ったイベントを別のクライアントで受け取る
このステップでは、一方のターミナルでコンシューマーを待機させ、もう一方から3件のレコードを送ります。Terminal サービスでは、ターミナルセッションを管理する tmux がシェルを保持しています。ブラウザーで2つ目の接続を開くと、最初は同じ継続中のセッションが表示されます。クライアントごとに独立したプロンプトを使えるよう、ターミナル B 用に名前付きのシェルセッションを作成します。
ターミナル A で、切り離された状態のセッションを作成します。-d は現在のターミナルをそのまま選択し、-s は新しいセッションに名前を付け、-c は開始ディレクトリを指定します。これはシェルを作成する操作で、Kafka ブローカーを追加する操作ではありません。
## Terminal A
tmux new-session -d -s kafka-b -c /home/labex/project
ターミナル A を開いたまま、同じ Terminal サービスを別のブラウザータブで開きます。新しいタブをターミナル B とし、その接続を用意したセッションに切り替えます。
## Terminal B
tmux switch-client -t kafka-b
切り替えた後、シェルだけが動作している状態で Ctrl-C を1回押し、未送信のターミナル入力を消します。セッション名を確認します。
## Terminal B
tmux display-message -p '#S'
kafka-b と表示されるはずです。ターミナル A は元のセッションを維持し、どちらのシェルからも同じ VM のファイルにアクセスできます。ブラウザータブを切り替える際も、この役割を保ってください。この演習ですでに名前付きセッションを作成している場合は、再作成せずに使い回します。
ターミナル B で次のコンシューマーを起動します。3件のレコードを読み取るまで待機し、その後は自動的に終了します。--from-beginning は新しいコンシューマーに対して保持されている先頭からの読み取りを要求し、--max-messages 3 は読み取り件数を制限し、--timeout-ms 30000 はアイドル状態で待つ時間を制限します。パイプで通常の出力を tee に渡し、画面への表示とローカルコピーの保存を同時に行います。
## Terminal B
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-first --from-beginning --max-messages 3 --timeout-ms 30000 | tee /home/labex/project/output/received-events.txt
ターミナル B を待機させたまま、ターミナル A に切り替え、30秒のアイドル制限内に次のバッチを送ります。何も送る前に時間切れになった場合は、ターミナル B でもう一度同じコンシューマーコマンドを起動してから、戻ってバッチを送ってください。すでにイベントを送った場合は、コンシューマーだけを再実行して保持された3件を確認します。プロデューサーのバッチを再送すると、重複するレコードが追加されます。
## Terminal A
printf '%s\n' 'order-1001 created' 'order-1002 created' 'order-1001 paid' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders-first
printf は引用符で囲んだ各テキスト値の後に改行を付けて出力します。パイプはその行をプロデューサーに渡し、このコンソールプロデューサーは1行を1つのイベントレコードとして扱います。printf が終了すると入力ストリームが閉じるため、プロデューサーは保留中の送信を完了して終了します。対話型のプロデューサープロンプトに入力する必要はありません。
ターミナル B に戻ります。データの3行は次のようになるはずです。
order-1001 created
order-1002 created
order-1001 paid
ターミナルには処理したメッセージ数やクライアントの診断メッセージも表示される場合があります。保存したファイルには3つのイベント値が入るはずです。order-1001 に関する2つの事実は別々のレコードです。paid イベントは、それより前の created イベントを上書きしていません。
通常、コンシューマーは3件のメッセージを受信するとプロンプトに戻ります。誤って上限のないクライアントを起動した場合は、そのターミナルで Ctrl-C を押すとそのクライアントが停止します。ブローカーの停止やトピックの削除は行われません。ここでは上限を設けているため、通常の手順では手動で中断せずに終了します。
新しいコンシューマーで保持されたイベントを読み取る
このステップでは、読み取りによってイベントが削除されたかを調べます。最初のコンシューマーは終了しています。新しい独立したコンシューマーが保持された先頭から読み始めると何が見えるか、予想してみてください。同じ3件のレコードが見えるはずです。
ターミナル B で、件数を制限した新しいコンシューマーを実行し、2回目の読み取りを別のファイルに保存します。
## Terminal B
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-first --from-beginning --max-messages 3 --timeout-ms 30000 | tee /home/labex/project/output/replayed-events.txt
この単一パーティション内では、同じ3つの値を同じ順序で取得できるはずです。別のバッチは送信していません。ターミナル A から、ローカルに保存した両方の観察結果を確認します。
## Terminal A
cat /home/labex/project/output/received-events.txt
## Terminal A
cat /home/labex/project/output/replayed-events.txt
どちらにも元の3件のイベントがすべて含まれるはずです。消費と保存期間は別の仕組みであり、読み取ってもレコードはトピックから削除されません。この最初の実験では、新しいコンソールコンシューマーを使用しています。後の演習では、名前付きのコンシューマーグループが読み取り位置を保存する方法と、先頭から読むオプションが、すでにコミットされた有効な位置を上書きしない理由を学びます。
短時間動作する両方のクライアントが終了した後も、Kafka のイベント履歴は利用できました。次はトピックのプロパティを調べ、保持の設定がコンシューマーの進行状況とは独立して保存方針に影響することを確認します。
まとめ
Kafka のイベント配信の役割を理解し、パーティションが1つのトピックを作成して、別々のプロデューサーとコンシューマーが3件の注文イベントを送受信する様子を観察しました。最初のコンシューマーが終了した後、新しいコンシューマーで保持された同じ履歴を読み取り、読み取りと削除が異なる操作であることを確かめました。次は、トピックのパーティション、レプリカ、保持に関するプロパティを調べます。
