介绍
一家小画廊让参观者为最喜欢的展位投票。发送方可能重试同一个事件，因此新增一条 Kafka 记录并不总是表示多了一票。你将先观察正常的两票样例，再观察真实存储在另一位置的重试，最后重建画廊的票数。
你应已掌握终端路径、Python 函数/循环/字典、JSON，以及 Apache Kafka for Beginners 中的 Kafka 键、分区、偏移量、消费组和 Python 客户端基础。本实验新增的概念是业务事件身份。环境已准备完整的原生读取程序，让你专注于计数规则；下一个引导实验再构建该读取程序。请使用 Terminal 和已安装的 bigdata-python 环境。作者估计需要 40–50 分钟，包含阅读和普通错误处理，并非初学者实测用时。
本实验只处理已经交付的有限批次，不涉及投票人资格系统、连续数据流、崩溃恢复保证或端到端恰好一次处理承诺。每个实验均独立初始化。
观察存储位置不同的一次重试
本步骤将比较两次普通投票与一个重试事件。请从 Terminal 开始。工作区已提供工具和展位目录。只编辑 reducer.py、replay.py 以及可选的 helpers/ 下的 Python 辅助程序；派生文件放在 output/ 下。保留所提供的工具/数据、冻结的源主题和已暂停的展示应用消费组。独立的演示主题只允许进行下面这一次重试操作。
cd /home/labex/project
-B 可避免这些 Python 命令生成字节码缓存文件。只读的 inspect 工具打印真实 broker 记录的封装信息：主题、分区、偏移量、键和解码后的值。末尾的数量表示已存储的应用记录条数。先查看小型演示：
bigdata-python -B vote_tool.py inspect demo
预期看到两条记录。它们的值遵循以下模式：event_id、booth 和 choice 都是非空字符串；choice 必须恰为 favourite；booth 必须出现在所选目录中。本课程的选票模式不允许额外字段。键是展位名称，用于路由；event_id 标识一张业务选票。这里，demo-1 投给 oak，demo-2 投给 fern：两条记录、两张逻辑选票。
发送之前，先预测两个结果：会存储多少条物理记录？如果 demo-1 仍是一张业务选票，各展位的计数会是多少？然后验证你的预测。
重试命令将完全相同的 demo-1 值发送给真实 broker，并等待交付确认。发送前会检查现有演示内容，因此在预期重试已存在时再次运行，不会追加第四条记录：
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
现在有三条记录。demo-1 在两个不同偏移量上出现，值的内容完全相同。这些位置标识两条已存储记录，而重复的事件 ID 仍只标识一票。正确的展位总票数依然是 oak=1、fern=1、reed=0。重新读取现有记录与再次存储一次重试是两种不同情况；构建计数时，两者都需要业务身份规则。
冻结的课程批次采用显式分区放置，因此相同的展位键可能出现在不同分区中。比较记录时，要读取真实元数据。
在所选课程批次中，事件 ID 的身份范围覆盖所有分区，并非仅限于某一分区或展位。即使其他每个字段都相同，两个不同 ID 仍计为两票。所提供的业务接受这些事件；它并不判断一个真实的人可以投几票。
如果工具提示批次未知，请列出已准备的名称，本次观察使用 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
shell 用 $? 保存紧邻的上一条命令的退出状态：零表示成功，非零表示失败。在其他命令覆盖它之前立即显示：
printf '%s\n' "$?"
summarize(values, booths) 对有效输入必须返回含有 votes 和 counts 的字典，对无效输入则引发普通异常。保持模块可调用；不要从中终止 Python 解释器。检查器必须收到真实返回值，而不只是解释器退出状态。
在 nano 中编辑现有函数
在 Terminal 的文本编辑器中打开可编辑文件：
nano reducer.py
顶栏显示文件名。用方向键移动，Backspace/Delete 修正字符。Python 通过缩进组织语句：函数内部使用四个空格，循环内部使用八个空格，每层嵌套条件再增加四个。保留已有的 def summarize(values, booths): 行。替换旧逻辑时，删除原来说明故意按记录计数的注释。
下面的简短 Python 代码块是这个函数内部相互连接的编辑内容，不是在 shell 中运行的命令，也不是各自独立的程序。完成所有编辑后再运行检查器。保留起始代码中有效的部分，不要替换整个文件。
首先，保留已有的零计数初始化，在紧接着的下方、循环之前添加 seen。这个以 event_id 为索引的字典将保存每张已接受业务选票的完整值：
counts = {booth: 0 for booth in booths}
seen = {}
接着，保留唯一已有的 for value in values: 行，将原来的累加替换为以下验证。代码块展示这条已有的 for 行作为上下文；不要再添加第二个循环。set(value) 收集字段名，isinstance 检查类型，any(...) 判断是否至少有一个字段值无效。第一个条件要求值是字典，恰好含三个非空字符串字段，且选择为允许值。尝试为展位累加之前先检查目录。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 {"votes": list(seen.values()), "counts": counts}
保存并退出。nano 底部标签中的 ^ 表示按住 Ctrl：按 Ctrl+O，再按 Enter 确认 reducer.py，然后按 Ctrl+X 返回 shell。这些按键保存当前文件，不会运行 Python。回读实际保存的内容：
cat reducer.py
检查每次函数调用只初始化一次 seen，模式/目录检查先于计数，只有首次出现的 ID 才累加，返回语句位于循环外。随后立即测试完成的函数：
bigdata-python -B check_rules.py
预期看到 Rules pass，涵盖身份、不同 ID、冲突、模式、零票展位和重新排序的输入。只以展位为键的字典会合并不同投票事件；以分区/偏移量为键的字典会保留全部重试。你的字典表示的是业务身份。
遇到缩进、语法或名称错误时，阅读报告的文件名和行号，再次运行 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
预期有五个逻辑选票 ID，展位计数如下：
Booth Votes
fern 2
oak 3
reed 0
快照还保留了全部八条物理封装记录，因此你能区分源记录与五张逻辑选票。reed 仍以零票保留。使用已准备的写入器时，发布前的循环或验证错误会保留上一份快照；入口不会自动回滚其他写入方式。每次编辑程序后，都应重新运行公共入口，再检查完成情况。
原展示应用已暂停，但保留了消费组进度。你可以描述该组，但不得加入、重置/删除它，也不得向它提交偏移量。不要编辑/删除源主题或增加分区，只重建自己的派生输出。检查会观察原主题身份、完整的分区/副本拓扑、真实记录以及展示应用保存的进度，不承诺保护任意 broker 配置。
你现在得到了有用的展位快照，而不只是交付记录。下一个实验将解释并构建完整的独立原生读取程序，包括为什么一次空轮询不代表批次已完成。
总结
你观察了真实存储的重试，区分了全局业务身份与 Kafka 记录位置，并拒绝了同 ID 内容冲突。真实画廊快照保留完整的物理来源，同时报告逻辑选票和每个展位的票数，包括零票。来源和已暂停的展示应用消费组仍然可用。
