はじめに
作成、支払い、発送など、複数のイベントで同じ注文の出来事を表せます。関連するレコードを一貫して振り分ければ、コンシューマーは1つのパーティション内でその注文の流れを追えます。レコードキーは、振り分けに使う安定した入力値です。個々のイベントの発生を識別する event_id とは異なります。
3つの注文のイベントを交互に含むデータを、3つのパーティションを持つトピックに送信し、実際のキーと位置を表示して、既存の注文のイベントをさらに1件追加します。オフセットは、1つのパーティション内でのレコードの位置です。トピック全体の行番号ではなく、異なるパーティションのレコード間には、保証された単一の全体順序はありません。
ブローカーと入力用のサンプルデータは独立して用意されています。基本的なターミナル操作ができれば十分です。イベント値には読みやすい JSON テキストを使いますが、Python、JSON のプログラミング、クラスターの設定、以前のトピックの状態は不要です。実験中はパーティション数とクライアントの振り分け方式を固定してください。Terminal を使用します。
注文キーを付けて交互に並ぶイベントを送信する
このステップでは、完全なイベントを値に保持したまま、安定した注文識別子をキーとして送ります。用意された5件のバッチを確認します。
cd /home/labex/project
cat /home/labex/project/data/keyed-orders.txt
各行は key|value という形式です。キーは注文識別子です。JSON 値では波括弧の中に名前付きフィールドを記述します。event_id は一意で、order_id はキーと一致し、sequence はその注文の業務上の順番、event_type は出来事の種類を表します。JSON テキストは値の形式であり、Kafka が自動的にテーブルのスキーマとして解釈するわけではありません。
パーティションが3つ、レプリカが1つのトピックを作成します。
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-keyed --partitions 3 --replication-factor 1
プロデューサーの --reader-property オプションは、入力行の解釈方法を設定します。parse.key=true と key.separator=| により、入力行を区切り文字の位置で分割します。区切り文字自体はキーにも値にも含まれません。シェルの < はファイルを標準入力として渡し、プロデューサーは有限の入力を送信すると終了します。
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders-keyed --reader-property parse.key=true --reader-property 'key.separator=|' < /home/labex/project/data/keyed-orders.txt
ファイルでは注文が交互に並ぶため、order-2001 のイベントは入力内で隣り合っていません。同じキーの振り分け方式を使い、パーティション数を変えなければ、等しいキーは同じパーティションに送られます。異なるキーが1つのパーティションを共有する場合もあり、注文が3つあっても、3つの異なるパーティション番号に分かれるとは限りません。
この5件のバッチは1回だけ送信してください。コンシューマーの読み取りを繰り返すのは確認操作ですが、このプロデューサーコマンドを繰り返すと、さらに5件が追加されます。送信後の状態から再開する場合は、ファイルを再送せず、コンシューマーによる確認に進みます。
パーティション番号を手動で選ぶ代わりに、振り分けの入力を指定しました。次は、振り分け先を推測せずに、実際の結果を読み取ります。
キー、パーティション、オフセットをまとめて調べる
このステップでは、各値と一緒にレコードのメタデータを表示します。コンソールコンシューマーの --formatter-property オプションは表示方法を制御し、通常の出力にキー、パーティション、オフセットを追加します。保存されたイベントの内容は変えません。新しいコンシューマーでの先頭からの読み取り、5件の上限、アイドル時のタイムアウトにより、この有限の確認操作を繰り返せます。出力をローカルに保存します。
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-keyed --from-beginning --max-messages 5 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/keyed-first.txt
cat /home/labex/project/output/keyed-first.txt
データの各行には、実際の数値と完全な JSON 値を使った、おおよそ次の形のフィールドが表示されるはずです。
Partition:... Offset:... order-... {"event_id":"...",...}
区切りはタブです。k1 と k3 を探してください。どちらもキーは order-2001 で、同じパーティションにあり、k3 のオフセットがより大きいはずです。同様に、k2 と k4 は order-2002 のパーティションを共有するはずです。この2つの注文が別々のパーティションに分かれることは条件にしないでください。
オフセットはキーではなくパーティションに属します。他のキーが同じパーティションを共有していると、ある注文の連続するイベントのオフセットには飛びが生じる場合があります。ペイロードの sequence フィールドはアプリケーションの値であり、Kafka がオフセットとして生成したものではありません。
コンシューマーは異なるパーティションのレコードを交互に表示する場合があるため、表示全体の順序はプロデューサーのファイルと一致するとは限りません。出力を全体で順序付けられた1つのリストとして扱わず、キー、パーティション、増加するオフセットを使って関連イベントを確認してください。
既存の注文にイベントを追加する
このステップでは、プロデューサーをもう一度起動し、同じ振り分け規則を試します。order-2001 の shipped イベントがどうなるか予想してください。この注文の既存のパーティションに送られ、それより前のレコードに続くはずです。正確なオフセットは、そのパーティションを共有する他のレコードの有無によって決まります。
最初の Kafka 演習で使った printf の形式で、完全なキー付きの1行を、同じ設定のコンソールプロデューサーに送ります。
printf '%s\n' 'order-2001|{"event_id":"k6","order_id":"order-2001","sequence":3,"event_type":"shipped"}' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders-keyed --reader-property parse.key=true --reader-property 'key.separator=|'
保持された6件すべてを読み取り、新しい観察結果として保存します。
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-keyed --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/keyed-with-shipped.txt
cat /home/labex/project/output/keyed-with-shipped.txt
order-2001 と sequence 3 を持つ k6 を探します。パーティションを k1、k3 と比較し、オフセットもそれぞれと比較してください。パーティション内のオフセット順に見ると、注文の流れは created、paid、shipped のままであるはずです。特定のパーティション番号である必要はありません。
元の5件の観察結果は keyed-first.txt に残っています。新しいプロデューサーは以前のイベントを置き換えず、イベントを追加しました。トピックのパーティション数や振り分け方式は振り分け先に影響するため、この実験ではどちらも変更していません。後から使うクライアントでも、実装間で同じ対応付けが必要なら、互換性のある振り分け方式を選ぶ必要があります。
イベント全体の順序や、注文ごとに1つの専用パーティションがあることを仮定せず、キーごとの安定した振り分けとパーティション内の順序を裏付ける結果が得られました。次のチャレンジでは、トピック、保持期間、キー付き送信のスキルを新しいバッチに適用します。
まとめ
注文キーを指定し、実際のパーティションとオフセットを調べ、同じ振り分け規則のもとで追加したイベントが、その注文の以前のレコードに続くことを確認しました。全体の順序を仮定せず、イベント ID、注文キー、ペイロードの順番、パーティションのオフセットを区別しました。次の新しいバッチのチャレンジでは、これらのスキルをトピックの保持設定と組み合わせます。
