はじめに
スタジオでは投票履歴が保持されていますが、元の表示アプリケーションは一時停止中です。その進捗を巻き戻したりログを変更したりせず、新しい票数を求める必要があります。新しいコンシューマーが固定済みバッチをまだ受信していない場合でも、ポーリングで一時的にメッセージが返らないことがあります。
ネイティブなリプレイループを構築し、各パーティションの終了位置をそれぞれ取得して、有限バッチ全体を読んだ後にだけ結果を書き出します。前のラボの論理的な識別関数は、この独立した VM に用意されています。そこで使った Python と Kafka の基礎が前提です。Terminal を使用してください。作者の見積もりは読解と通常の間違いへの対処を含めて 50～60 分で、初心者による計時試験ではありません。
これは固定済みバッチの再構築です。演習中にこれらのトピックへ追加のメッセージは公開されず、トランザクションを実装する課題でもありません。信頼できるクラッシュ復旧や、エンドツーエンドで厳密に 1 回だけ処理することも保証しません。
各パーティションの境界を取得し境界まで読む
このステップでは、準備済みの外側のプログラム内に、完全で独立したネイティブ読み取り処理を実装します。Terminal から始めます。replay.py と、必要なら reducer.py または helpers/ 配下の Python 補助プログラムを編集し、派生ファイルは output/ 配下に書きます。提供されたツールとデータ、元のトピック、一時停止中の表示アプリケーションのグループは保持します。
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
読み取り専用の説明には、実際の studio-votes のパーティション ID、保持開始位置 low、現在の固定済み末尾、レプリカが表示されます。パーティションは順序のあるログであり、別の VM ではありません。末尾は保持されたバッチの次の位置で、数える対象のレコードではありません。どのパーティションに読み取るレコードが必要か予想します。空のパーティションは low == high で、すでに完了しています。
オフセットの幅はレコード件数ではありません。Kafka は内部位置、制御用の位置、削除済みの位置を飛ばすことがあります。このソースには実際のネイティブな位置の欠番があります。補って作るべき票が欠けているわけではありません。実レコード全体の情報を確認し、その実際の位置を説明と比較します。
bigdata-python -B vote_tool.py inspect main
再構築の前に、前提コースのコンシューマーグループツールで、一時停止中の表示アプリケーションの保存済み位置を読みます。
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
後で比較するため、各 CURRENT-OFFSET を記録します。アクティブなメンバーがいないのは正常です。表示アプリケーションは一時停止中ですが、保存済みの進捗があります。ここでの LAG はログ位置の差であり、論理的な票数ではありません。独立したコンシューマーは、このグループの進捗を動かしてはいけません。
準備済みの外側のプログラムを読む
cat replay.py
インポート、arguments()、クライアント設定、エラー報告、finally と close は提供済みです。中心部分は意図的に NotImplementedError を送出します。パーティションを読んだり、サンプルの解答を書き出したりはしません。前のラボの正しい識別関数は、別の reducer.py に用意されています。
クライアントは、選択した bootstrap.servers のアドレスと、独立した group.id を使います。allow.auto.create.topics は false です。自動コミットとオフセットの自動保存は無効です。assign は、動的なグループ割り当てを使わず、明示的な開始位置を与えます。この例は明示的なコミット呼び出しを行いません。同等の独立グループ実装が自分の進捗を保存することは可能ですが、元の表示アプリケーションのグループへの参加、コミット、リセット、削除は決して行ってはいけません。
準備済みの vote_support.py は、トピック、カタログ、出力、broker、グループを扱う arguments() と、完全な結果を書き込む finish(...) を提供します。パーティションを発見したり、票を取得したりはしません。finish は、結果の書き込みや置き換えの前に識別規則を呼び出します。finally は、成功時も失敗時もコンシューマーを閉じます。この外側の構造を残してください。
短いつながった部分でネイティブ処理を実装する
前のラボのエディター操作を使います。
nano replay.py
インポート、引数解析、Consumer の設定、try:、except、finally の各部分は残します。try: の内側にある二つの説明コメントと、未完成の中心部分の raise NotImplementedError(...) だけを削除します。矢印キーで位置を合わせ、Backspace/Delete で編集します。Ctrl+K は現在の行全体を切り取れるため、置き換えるつもりの行にいるときだけ使います。
空いた場所に、次のつながった部分を挿入します。これらは一つの中心処理を構成します。個別のブロックをシェルで実行したり、作りかけのループを保存して実行したりしないでください。すべての部分を完成させ、提供された外側のエラー処理とクローズの枠組みを残してから、保存して実行します。四つのスペースで try: の内側、さらに四つ追加するとループや条件の内側になります。表示された相対的なインデントは、Python の意味の一部です。
まず、完全な実パーティション集合を発見します。対象トピックを指定せずに list_topics を呼ぶことで、不明な名前に対するメタデータ要求を避けます。選択した項目を使う前に、利用可能か確認します。発見した ID をソートすると、この例の割り当てリストが安定します。グローバルなイベント順序を作るわけではありません。
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
発見した各パーティションで、get_watermark_offsets(..., timeout=10) を一度要求します。TopicPartition は、その要求のトピックとパーティションを識別します。保持開始位置 low と固定した high の両方を保存し、診断用の次の位置を low で初期化します。ポーリングループ内で high を更新してはいけません。伸び続ける末尾を追うと、有限の再集計ではなくなります。この演習では、固定済みソースへの追加や変更は禁止されています。
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
読み取るべき保持位置があるパーティションだけで pending を作ります。assign は、空のパーティションの low も含め、明示的な開始オフセットのリストを受け取ります。実行ごとに新しい物理レコードリストを初期化します。time.monotonic() はシステム時刻の調整とは独立して経過時間を測り、この期限が未完了の読み取り時間を制限します。
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
ループを開始します。poll(0.2) は、メッセージ、エラーやイベントを含むメッセージ、または待機時間切れで None を返すことがあります。なぜ None を完了の意味にできないのか予想します。クライアントが、固定済みレコードをまだ受信していないだけかもしれません。全体の期限内で待機を続けます。設定ではパーティション EOF イベントが有効です。_PARTITION_EOF は想定された終了イベントで、他のエラーは実行を失敗させます。これにより、最後の内部位置を越えるネイティブな進捗を扱えます。
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
同じ if message is not None ブロックを、次の elif で続けます。十二個のスペースで if message.error(): とそろえます。未完了のパーティションにあり、固定した high より手前の通常メッセージだけを、レコード全体の情報として保存します。物理的な再送も保持します。論理的な識別は、後で summarize が扱います。キーと値のバイト列を UTF-8 でデコードし、その後 json.loads で値を辞書にします。
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
今回の poll を処理した後、残っている各パーティションについて consumer.position(...) を問い合わせます。これはコンシューマーの次の位置であり、どのグループのコミット済み進捗とも異なります。推測したレコード件数ではなく、取得済みの high と比較します。完了したパーティションだけを pending から取り除きます。for position 行は八つのスペースで、while の内側、メッセージ条件の外側に置きます。そのため、何も返らない poll の後でも位置を確認できます。最後に、ループが完了した後、四つのスペースで finish を呼び出します。
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
各パーティションに固定した境界ができました。空のパーティションに票は不要で、欠番の位置に架空のレコードは不要であり、None でも読み取りを止めませんでした。未完了の各パーティションの次の位置が high に達してから、初めて完全なスナップショットを書き出します。
Ctrl+O、Enter で replay.py を確定して保存し、Ctrl+X で終了します。保存したプログラムを確認します。
cat replay.py
六つの部分すべてが提供済みの try の内側にあること、レコード全体の情報の処理が通常メッセージの分岐内にあること、位置確認がループ内にあること、finish がループの外側にあることを確認します。直ちに選択したネイティブバッチに対して実行します。
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
物理レコード八件、論理的な票五票、cedar=3、maple=2、willow=0 が期待されます。公開エントリーポイントは、まず識別関数を使い捨ての例でチェックし、その後、実際のネイティブプログラムと補助プログラムのツリーが受信した broker データを記録します。別の完全な業務比較が実際の結果をチェックします。スナップショットが申告する境界や位置は診断用であり、ネイティブ処理の独立した証明ではありません。
成功する各呼び出しは、その実行中に結果を実際に書き込む必要があります。直接書くか、一時ファイルに書いてから選択した出力へ名前変更します。再実行で同じバイト列になることは有効で、補助の子プロセスが書いてもかまいません。古いファイルを残すだけ、そのタイムスタンプを更新するだけ、または完了マーカーを書くことは、結果の書き出しにはなりません。提供された finish は、学んだ一時ファイルと置き換えの方法を使います。その準備済みの書き込み処理では、書き出し前のループや検証のエラー時に、以前のスナップショットを利用可能なまま残します。エントリーポイントは、任意のその場の上書きをロールバックするものではありません。
構文、名前、インデントのエラーには、報告された行を使って nano replay.py を開き直し、動作する部分を残してその箇所を修正します。保存して終了し、cat replay.py で確認してから公開エントリーポイントを再実行します。リプレイ未完了によるタイムアウトには、選択したトピック引数と、pending、EOF、位置の処理を確認します。元のグループを巻き戻す理由にはなりません。pwd と ls replay.py reducer.py で、入力を間違えたパスを確認します。ローカルのプログラムエラーのために、setup の再実行やソースレコードの変更をする必要はありません。
最初から再構築し別のバッチを使う
このステップでは、再実行が同じ有限の結果を再構築することを確認し、別のカタログとパーティション分布を使います。引き続き Terminal の /home/labex/project で作業します。
用意された書き込み処理は、JSON 保存前に、物理レコード全体の情報をパーティション/オフセット順、論理的な票を ID 順に並べます。これにより、グローバルなイベント順序を保証せずに、同じ完全なスナップショットを安定させられます。sha256sum は内容のフィンガープリントを計算します。> で出力ディレクトリに保存し、後から -c で指定した実ファイルと保存済みフィンガープリントを比較します。
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
output/main-snapshot.json: OK が期待されます。各呼び出しでは、records、seen、カウントを空の状態から始めます。履歴の票を以前の合計に再び加えるのではなく、完全なスナップショットを置き換えます。消費によってレコードが削除されたり、一時停止中の表示アプリケーションが巻き戻されたりしたわけではありません。
再実行後、元の表示アプリケーションの位置をもう一度読みます。
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
すべての CURRENT-OFFSET が、先ほど記録した一時停止中の表示アプリケーションの表と一致する必要があります。独立した再構築は、保持されたレコードを読んだものの、そのグループへのコミットも巻き戻しも行っていません。アクティブなメンバーがいないというメッセージも正常なままです。
check バッチを一覧で確認し、そのブースカタログを読みます。
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
このバッチはブース名が異なり、パーティションも 4 つではなく 3 つです。同じプログラムファイルを使ってください。引数の解析で実際のトピックとカタログを選び、発見処理で完全な実パーティション集合を取得します。
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
物理レコード 6 件、論理的な票 4 票、amber=2、violet=2、moss=0 が期待されます。最初のカタログをハードコードしたり、最初のトピックのレコード件数で停止したりしても、このバッチを正しく表せません。
最初の実ファイルに戻ります。
bigdata-python -B read_snapshot.py output/main-snapshot.json
スタジオの結果がそのまま残っています。読み取りツールは指定されたファイルを直接開き、Kafka から何も補いません。ファイルがないと報告されたら、vote_tool.py list とファイル名を照らし合わせてパスを直してください。対応する結果の再構築が本当に必要なときだけ、そのバッチを再構築します。
衝突を拒否し有効なスナップショットを保つ
このステップでは、有用な結果を置き換えずにデータの誤りを観察します。引き続き Terminal の /home/labex/project で作業します。一覧の conflict トピックには、同じ ID に対して異なる有効なブース内容を持つレコードが意図的に含まれています。出力先は既存の主スナップショットです。どちらの値も正解として認められていないため、成功した結果としての書き出しを拒否する必要があります。
前のステップのフィンガープリントは、有効な主ファイルを示しています。同じプログラムを衝突バッチに対して実行します。次のコマンドは失敗するのが想定どおりです。
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
ゼロ以外の終了ステータスと、実際の ID を示す Conflicting event_id が期待されます。規則は、用意された書き込み処理が結果を作成/置換する前に実行されます。ここでの .pending/置き換え方式は、完全な結果を書き出すための方法であり、クラッシュ復旧や耐久性の保証ではありません。
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
有効な主ファイルは引き続きフィンガープリントと一致し、スタジオの元の票数を持つ必要があります。失敗した衝突実行は、成功した main/check 実行とは別に記録されます。衝突ソースを編集したり、先に届いたイベントを適当に選んだりしないでください。逆順やパーティションごとの順次走査でも、同じ衝突を拒否する必要があります。
普通のバッチ名の入力ミスなら、エントリーポイントはリプレイ開始前に利用可能な名前を表示します。vote_tool.py list で確認し、正しい名前で実行し直してください。ローカルのプログラムエラーなら、ファイルを修正し、影響を受けた有効なバッチを再実行します。どのようなソースプログラムの編集後でも、両方の有効なバッチを再実行して結果を現在のファイルに対応させ、その後に衝突のセルフチェックを繰り返してください。
元の表示アプリケーションのグループは、一時停止したままにします。参加、コミット、リセット、削除は禁止です。リセット後に同じ最終オフセットへ戻しても、独立したリプレイの範囲外です。機能チェックは、プログラムツリーのネイティブ受信と完全な実出力を組み合わせますが、計算の因果関係や、管理者が環境を偽装することへの耐性は証明しません。
まとめ
各パーティションの実境界を取得し、一時的に結果がないポーリング、空のパーティション、欠番の位置があっても、有限ログ全体を読みました。同じネイティブプログラムが、古い合計を累積せず 2 つのカタログの結果を再構築しました。内容が衝突する ID は失敗となり、有用な主スナップショットと元のコンシューマーの進捗は保持されました。
