はじめに
小さなギャラリーで、来場者がお気に入りのブースに投票します。送信側が同じイベントを再送することがあるため、Kafka のレコードが増えても、必ずしも票が増えたとは限りません。まず通常の 2 票の例を観察し、次に別の保存位置に実際に格納された再送を確認して、最後にギャラリーの票数を再構築します。
端末のパス、Python の関数・ループ・辞書、JSON、および Apache Kafka for Beginners で扱う Kafka のキー、パーティション、オフセット、コンシューマーグループ、Python クライアントの基礎を前提とします。このラボで追加するのは、業務イベントの識別情報です。票数の規則に集中できるよう、完全なネイティブ読み取りプログラムは用意されています。次のガイド付きラボで、その読み取りプログラムを構築します。Terminal とインストール済みの bigdata-python 環境を使用してください。作者による所要時間の見積もりは 40～50 分で、読解と通常の間違いへの対処を含みます。初心者の実測時間ではありません。
扱うのは、すでに届けられた有限のバッチだけです。投票資格を確認する仕組み、継続的なストリーム、クラッシュからの復旧保証、エンドツーエンドで厳密に 1 回だけ処理するという保証はありません。各ラボは独立した初期状態から始まります。
別の保存位置にある再送を観察する
このステップでは、通常の 2 票と再送されたイベントを比較します。Terminal から始めてください。作業領域には、ツールとカタログが用意されています。編集してよいのは reducer.py、replay.py、必要に応じて helpers/ 配下の Python 補助プログラムだけです。派生ファイルは output/ 配下に置きます。提供されたツールとデータ、固定済みのソーストピック、一時停止中の表示アプリケーションのグループは維持してください。独立したデモ用トピックで許可されている操作は、次の再送だけです。
cd /home/labex/project
-B は、これらの Python コマンドによるバイトコードのキャッシュファイル作成を防ぎます。読み取り専用の inspect ツールは、実際の broker のレコード全体の情報、つまりトピック、パーティション、オフセット、キー、デコードした値を表示します。最後の件数は、保存されたアプリケーションレコードの数です。小さなデモを調べます。
bigdata-python -B vote_tool.py inspect demo
2 件のレコードが表示されるはずです。値のスキーマは次のとおりです。event_id、booth、choice は空でない文字列で、choice は正確に favourite、booth は選択したカタログに含まれる値です。このコースの投票スキーマに余分なフィールドはありません。キーはブース名で、ルーティングに使います。event_id は業務上の 1 票を識別します。ここでは、demo-1 が oak、demo-2 が fern に投票しています。レコードは 2 件、論理的な票数も 2 票です。
送信前に二つの結果を予想します。保存される物理レコードは何件になるでしょうか。また、demo-1 が依然として業務上の 1 票であるなら、ブース別の票数はどうなるでしょうか。その後、予想を確かめます。
再送コマンドは、まったく同じ demo-1 の完全な値を実際の broker に送り、配送確認を待ちます。送信前に既存のデモを確認するので、期待する再送がすでにある状態で再実行しても、4 件目のレコードは追加されません。
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
これでレコードは 3 件になりました。demo-1 は異なる 2 つのオフセットにあり、値の内容は同一です。これらの位置が識別するのは 2 件の保存レコードですが、重複したイベント ID が識別する票は依然として 1 票です。正しいブース別合計は oak=1、fern=1、reed=0 のままです。既存のレコードをもう一度読むことと、再送を別途保存することは異なる状況です。どちらも、票数を構築する際には業務上の識別規則が必要です。
このコースの固定済みバッチは、パーティションを明示して配置しています。そのため、同じブースキーが異なるパーティションに現れることがあります。レコードを比較するときは、実際のメタデータを読んでください。
選択したコースのバッチ内では、イベント ID は、1 つのパーティションやブースだけでなく、すべてのパーティションを通じて共通の識別情報です。それ以外のフィールドがすべて同じでも、異なる 2 つの ID は別々の 2 票として数えます。提供された業務の規則は、そのようなイベントを受け入れます。実際の人が何票投じてよいかを判断するものではありません。
ツールが不明なバッチを報告した場合は、用意された名前を一覧表示し、この観察では demo を使用します。
bigdata-python -B vote_tool.py list
入力ミスによって broker がリセットされたわけではありません。引数を直し、読み取り専用の確認を繰り返してください。表示された票数を直すために、任意のイベントを再送しないでください。
論理的な識別情報を保ち内容の衝突を拒否する
このステップでは、準備済みの Python 関数の内部に業務規則を実装します。引き続き Terminal の /home/labex/project で作業します。現在のファイルを読みます。
cat reducer.py
カタログのカウントを初期化した後、入力値ごとに加算しています。デモの三件のレコードを与えるとどうなるか予想してください。二件目として保存された demo-1 でも票が増えるはずです。正しい規則は、その ID を一回だけ数え、異なる ID は別々に保持する必要があります。
準備済みの規則チェッカーは、関数の正確なコピーを使い、順序を変えた値や異なるブース名を含む使い捨ての例を実行します。実際の Kafka ログは変更しません。現在の単純な関数を実行します。このコマンドは、規則のメッセージを表示して失敗するのが想定どおりです。
bigdata-python -B check_rules.py
シェルは、直前のコマンドの終了ステータスを $? に保存します。ゼロなら成功、ゼロ以外なら失敗です。別のコマンドに置き換えられる前に、直ちに表示します。
printf '%s\n' "$?"
summarize(values, booths) は、有効な入力では votes と counts を含む辞書を返し、無効な入力では通常の例外を送出しなければなりません。モジュールは呼び出し可能なままにし、そこから Python インタープリターを終了しないでください。チェッカーが受け取る必要があるのは、単なるインタープリターの終了ステータスではなく、実際の戻り値です。
nano で既存の関数を編集する
Terminal のテキストエディターで、編集可能なファイルを開きます。
nano reducer.py
上部のバーにファイル名が表示されます。矢印キーで移動し、Backspace/Delete で文字を修正します。Python はインデントで文をまとめます。関数内は四つ、ループ内は八つのスペースを使い、条件を入れ子にするたびにさらに四つ追加します。既存の def summarize(values, booths): 行を残します。意図的にレコード件数を数えるという古いコメントは、そのロジックを置き換えるときに削除します。
以下の短い Python ブロックは、この一つの関数の内側で行うつながった編集であり、シェルで実行するコマンドでも別々のプログラムでもありません。すべての編集を終えてからチェッカーを実行します。ファイル全体を置き換えず、起点のコードの有効な部分は残します。
まず、既存のゼロ件数の初期化を残し、ループの前、その直下に seen を追加します。event_id をキーとする辞書で、受け入れた各業務上の投票の完全な値を保持します。
counts = {booth: 0 for booth in booths}
seen = {}
次に、既存の for value in values: 一行を残し、元の加算処理を以下の検証に置き換えます。ブロック内の既存の for 行は前後関係を示すためのものです。二つ目のループを追加してはいけません。set(value) はフィールド名を集め、isinstance は型を確認し、any(...) は無効なフィールド値が少なくとも一つあるかを調べます。最初の条件は、承認された choice と、空でない文字列フィールドを正確に三つ持つ辞書を要求します。ブースの加算を試みる前に、カタログを確認します。for 行は四つ、その検証は八つのスペースにします。
for value in values:
if (not isinstance(value, dict)
or set(value) != {"event_id", "booth", "choice"}
or any(not isinstance(item, str) or not item for item in value.values())
or value["choice"] != "favourite"):
raise ValueError("Invalid vote schema")
if value["booth"] not in counts:
raise ValueError("Unknown booth: " + value["booth"])
カタログの検査の直後、引き続きループの内側に、識別処理を追加します。既知の ID では、デコード済みの値全体を比較します。内容が異なる場合は、その ID を示すエラーを送出しなければなりません。最初か最後の値を選ぶと、答えが走査順に依存してしまいます。同一の再送は continue に進み、再加算せず次の入力へ移ります。初めての ID は保存して数えます。
event_id = value["event_id"]
if event_id in seen:
if seen[event_id] != value:
raise ValueError("Conflicting event_id: " + event_id)
continue
seen[event_id] = dict(value)
counts[value["booth"]] += 1
dict(value) は受け入れた値を辞書として保持します。JSON の空白とフィールド順は業務上の識別情報ではありません。辞書のキーは、選択したバッチの全パーティションにわたって共通であり、ブース名でもパーティションとオフセットの組でもありません。同じブースと選択を持つ別の ID は、それぞれ初回 ID の加算まで進む必要があります。
最後に、全入力値を使っていた古い return を置き換えます。受け入れた一意な値と、カタログ全体のカウントを返します。この行は、ループの外側で関数の内側になる四つのスペースです。
return {"votes": list(seen.values()), "counts": counts}
保存して終了します。nano の下部の表示で ^ は Ctrl を押しながら操作する意味です。Ctrl+O を押し、Enter で reducer.py を確定し、Ctrl+X でシェルに戻ります。これらのキーは現在のファイルを保存し、Python を実行するものではありません。実際に保存した内容を読み返します。
cat reducer.py
seen が関数呼び出しごとに一度初期化されること、スキーマとカタログの検査が加算より前にあること、初めての ID だけが加算されること、return がループの外にあることを確認します。直ちに完成した関数を試します。
bigdata-python -B check_rules.py
識別情報、異なる ID、衝突、スキーマ、ゼロ票のブース、入力順の変更について Rules pass が表示されるはずです。ブースだけをキーにした辞書は別々の投票イベントをまとめ、パーティションとオフセットをキーにした辞書は再送をすべて残してしまいます。あなたの辞書は、業務上の識別情報を表します。
インデント、構文、名前のエラーでは、報告されたファイル名と行を読み、もう一度 nano reducer.py を実行し、矢印キーで該当部分を修正します。Ctrl+O、Enter で保存し、Ctrl+X で終了し、cat reducer.py で確認して規則チェックを再実行します。正しい部分は残します。関数全体の書き直しや VM のリセットは不要です。パスの食い違いは、次の読み取り専用コマンドで確認できます。
pwd
ls reducer.py replay.py
必要なら cd /home/labex/project で戻ります。次のステップでは、準備済みのネイティブ読み取りプログラムで、実際の Kafka レコードに規則を適用します。
実際のギャラリーログからスナップショットを書き出して読む
このステップでは、実際の固定済みギャラリーバッチに関数を適用します。引き続き Terminal の /home/labex/project で作業します。main バッチが選ぶソースと出力を一覧表示し、カタログを確認してください。
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main は gallery-votes を指します。複数のパーティション、実際に異なる保存位置にある同一 ID、他の内容が同じで ID だけが異なるイベント、空のパーティションを含みます。replay.py は用意されたネイティブ読み取りプログラムで、vote_support.py は引数解析とスナップショット書き込みを提供します。読み取りプログラムは、各実レコードの完全な情報に含まれる値を summarize 関数へ渡します。ローカルの投票入力ファイルから値を取得することはありません。
次の公開エントリーポイントを実行します。まず現在の識別関数を使い捨ての例でチェックし、現在のプログラムと補助の子プロセスを実行します。そのプロセスツリーのネイティブ読み取りの証拠を保存し、成功した結果を一覧に示された出力先へ書き込みます。
bigdata-python -B run_replay.py main
records=8, logical_votes=5 の書き出しが期待されます。エントリーポイントは、同じクライアントラベルの無関係なコンシューマーを信用するのではなく、このプログラムツリーが実際に broker データを受信したことを記録します。内部計算がその入力に基づいたことや、管理者による偽装への耐性を証明するものではありません。最終的な業務結果は、実際の完全なソースと別に比較されます。
成功する各実行は、その呼び出し中に結果を実際に書き出さなければなりません。ファイルデータを直接書くか、一時ファイルに書いてから結果ファイルへ名前変更します。補助の子プロセスが書いてもかまいません。同じバイト列を再び書き出すことは有効です。Kafka を読んで古いファイルを残すだけでは、新しい書き出しにはなりません。
実際の結果ファイルを読みます。この読み取りツールは、不足する結果を黙って取得、補完、修復することはありません。
bigdata-python -B read_snapshot.py output/main-snapshot.json
5 つの論理的な投票 ID と、次のブース別票数が期待されます。
Booth Votes
fern 2
oak 3
reed 0
スナップショットは 8 件すべての物理レコード全体の情報も保持するため、ソースレコードと 5 つの論理的な票を区別できます。reed はゼロ票で残ります。用意された書き込み処理では、書き出し前のループや検証のエラー時に、以前のスナップショットを利用可能なまま残します。エントリーポイントが他の書き込み方法を自動的にロールバックするわけではありません。プログラムを編集したら、完了チェックの前に公開エントリーポイントを再実行してください。
元の表示アプリケーションは、コンシューマーグループの進捗を保存して一時停止しています。グループの説明を読むことはできますが、参加、リセット/削除、そのグループへのオフセットのコミットは禁止です。ソーストピックを編集/削除したり、パーティションを追加したりしないでください。自分の派生出力だけを再構築します。チェックは、元のトピックの識別情報、完全なパーティション/レプリカ構成、実レコード、表示アプリケーションの保存済み進捗を確認します。任意の broker 設定を保護するという保証ではありません。
これで、単なる配送記録ではなく、役立つブースのスナップショットができました。次のラボでは、空のポーリング 1 回だけでバッチが終わったとはいえない理由を含め、完全な独立ネイティブ読み取りプログラムを説明し、構築します。
まとめ
実際に保存された再送を観察し、グローバルな業務上の識別情報を Kafka のレコード位置と区別して、同一 ID の内容衝突を拒否しました。実際のギャラリースナップショットは完全な物理ソースを保持しながら、論理的な票と、ゼロ票を含むすべてのブースを示します。ソースと、一時停止中の表示アプリケーションのグループもそのまま利用できます。
