Введение
Небольшая галерея предлагает посетителям проголосовать за любимый стенд. Отправитель может повторить отправку события, поэтому новая запись Kafka не всегда означает дополнительный голос. Сначала вы рассмотрите обычный пример с двумя голосами, затем настоящую повторную отправку в другой сохранённой позиции и, наконец, восстановите подсчёт голосов галереи.
Предполагается знание путей в терминале, функций/циклов/словарей Python, JSON, а также основ ключей, партиций, смещений, групп консьюмеров и клиента Python для Kafka из Apache Kafka for Beginners. В этом лабораторном занятии добавляется идентичность бизнес-события. Полный нативный код чтения уже подготовлен, чтобы вы сосредоточились на правиле подсчёта; в следующем пошаговом занятии вы построите этот код чтения. Используйте Terminal и установленную среду bigdata-python. Оценка автора в 40–50 минут включает чтение и обычные ошибки, но не является измерением времени работы начинающих.
Рассматривается только конечный, уже доставленный пакет. Здесь нет системы проверки права голосования, непрерывного потока, гарантии восстановления после аварии или сквозной обработки ровно один раз. Каждое занятие начинается независимо.
Рассмотрите повторную отправку в другой сохранённой позиции
На этом шаге вы сравните два обычных голоса с повторно отправленным событием. Начните в Terminal. В рабочем пространстве подготовлены инструменты и каталоги. Редактируйте только reducer.py, replay.py и при необходимости вспомогательные программы Python в helpers/; производные файлы размещайте в output/. Сохраните предоставленные инструменты/данные, зафиксированные исходные топики и группы приостановленных приложений отображения. В отдельном демонстрационном топике разрешено только одно действие: повторная отправка ниже.
cd /home/labex/project
-B запрещает этим командам Python создавать файлы кеша байткода. Инструмент только для чтения inspect выводит реальные оболочки записей брокера: топик, партицию, смещение, ключ и декодированное значение. Итоговое число означает количество сохранённых записей приложения. Рассмотрите небольшую демонстрацию:
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 настоящему брокеру и ожидает подтверждения доставки. Перед отправкой она проверяет существующую демонстрацию, поэтому повторный запуск после ожидаемой отправки не добавит четвёртую запись:
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
Опечатка не сбросила брокер. Исправьте аргумент и повторите проверку только для чтения. Не отправляйте произвольные события заново, чтобы исправить отображаемое число голосов.
Сохраните логические идентификаторы и отклоняйте противоречивое содержимое
На этом шаге вы реализуете бизнес-правило внутри подготовленной функции 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(...) определяет, некорректно ли хотя бы одно значение поля. Первое условие требует словарь ровно с тремя непустыми строковыми полями и разрешённым выбором. Проверьте каталог до попытки увеличить счётчик стенда. Оставьте четыре пробела перед строкой 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, чтобы вернуться в оболочку. Эти клавиши сохраняют текущий файл, а не запускают 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. Точка входа фиксирует реальное получение данных брокера этим деревом программы, а не доверяет постороннему консьюмеру с той же клиентской меткой. Она не доказывает, что внутреннее вычисление действительно опиралось на эти данные, и не защищает от подделки администратором. Конечный бизнес-результат отдельно сравнивается с настоящим полным источником.
Каждый успешный запуск должен действительно опубликовать результат во время этого вызова: напрямую записать данные файла либо записать временный файл и переименовать его в результат. Запись может выполнить вспомогательный дочерний процесс. Повторно публиковать идентичные байты допустимо; чтение Kafka с сохранением старого файла не является новой публикацией.
Прочитайте настоящий файл результата. Этот инструмент никогда незаметно не получает, не дополняет и не исправляет отсутствующие результаты:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Ожидаются пять ID логических голосов и такие счётчики стендов:
Booth Votes
fern 2
oak 3
reed 0
Снимок также сохраняет все восемь физических оболочек записей, позволяя отличить исходные записи от пяти логических голосов. reed остаётся с нулём голосов. При использовании подготовленного способа записи ошибки цикла или проверки до публикации оставляют предыдущий снимок доступным. Точка входа не откатывает другие способы записи автоматически. После любой правки программы повторно запускайте общедоступную точку входа, прежде чем проверять завершение.
Исходное приложение отображения приостановлено и сохраняет прогресс своей группы консьюмеров. Вы можете просматривать описание группы, но не вступать в неё, не сбрасывать/удалять её и не фиксировать для неё смещения. Не редактируйте/удаляйте исходный топик и не добавляйте партиции. Перестраивайте только собственный производный результат. Проверка наблюдает исходную идентичность топика, полную топологию партиций/реплик, настоящие записи и сохранённый прогресс приложения; она не обещает сохранение любой произвольной конфигурации брокера.
Теперь у вас полезный снимок стендов, а не только протокол доставки. Следующее занятие объясняет и строит полный независимый нативный код чтения, включая причину, по которой один пустой опрос ещё не означает завершение пакета.
Резюме
Вы наблюдали настоящую сохранённую повторную отправку, отделили глобальную бизнес-идентичность от позиций записей Kafka и отклонили противоречивое содержимое с одним ID. Ваш реальный снимок галереи сохраняет полный физический источник и показывает логические голоса и все стенды, включая нулевые. Источник и группа приостановленного приложения отображения остаются доступны.
