Введение
Студия сохранила историю голосования, пока исходное приложение отображения приостановлено. Вам нужен новый подсчёт без отката его прогресса и изменения журнала. Опрос может ненадолго не вернуть сообщения, хотя новый консьюмер ещё не получил зафиксированный пакет.
Вы построите нативный цикл повторного чтения, отдельно получите конечную позицию каждой партиции и опубликуете результат лишь после чтения полного конечного пакета. Функция логической идентичности из предыдущего занятия уже предоставлена в этой независимой VM. Нужны использованные там основы Python и Kafka. Используйте Terminal. Оценка автора в 50–60 минут включает чтение и обычные ошибки; это не хронометрированное испытание с начинающими.
Это восстановление зафиксированного пакета. Во время упражнения в эти топики не продолжают публиковаться сообщения, реализация транзакций не требуется, а надёжное восстановление после аварии и сквозная обработка ровно один раз не гарантируются.
Получите границу каждой партиции и читайте до неё
На этом шаге вы реализуете полное независимое нативное чтение внутри подготовленного внешнего каркаса программы. Начните в Terminal. Редактируйте replay.py и при необходимости reducer.py или вспомогательные программы Python в helpers/; производные файлы записывайте в output/. Сохраните предоставленные инструменты/данные, исходные топики и группы приостановленных приложений отображения:
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
Описание только для чтения показывает настоящие ID партиций studio-votes, удерживаемую границу 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/закрытие уже предоставлены. Основная часть намеренно возбуждает NotImplementedError; она не читает партиции и не публикует пример ответа. Правильная функция идентичности из предыдущего занятия отдельно подготовлена в reducer.py.
Клиент использует выбранный адрес bootstrap.servers и независимую group.id. allow.auto.create.topics имеет значение false. Автоматическая фиксация и автоматическое сохранение смещений отключены. assign задаст явные начальные позиции без динамического распределения группой; этот пример явно не вызывает фиксацию. Эквивалентная реализация с независимой группой может сохранять собственный прогресс, но никогда не должна вступать в исходную группу отображения, фиксировать для неё смещения, сбрасывать или удалять её.
Подготовленный vote_support.py предоставляет 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")),
})
После обработки этого опроса запросите consumer.position(...) для каждой оставшейся партиции. Это следующая позиция консьюмера, отличная от зафиксированного прогресса любой группы. Сравните каждую с полученной high, а не с предполагаемым числом записей. Удаляйте из pending только завершённые партиции. У строки for position восемь пробелов: она внутри while, но вне условия сообщения, поэтому даже после пустого опроса можно проверить позиции. Наконец, вызовите 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. Общедоступная точка входа сначала проверяет функцию идентичности на одноразовых примерах, затем фиксирует данные брокера, полученные настоящим нативным деревом программы/помощников. Отдельное полное бизнес-сравнение проверяет настоящий результат. Указанные в снимке границы/позиции служат диагностике, а не независимому доказательству нативной обработки.
Каждый успешный вызов должен действительно записать свой результат: напрямую либо через записанный временный файл, переименованный в выбранный выходной файл. Идентичные байты при повторе допустимы, запись могут выполнять вспомогательные дочерние процессы. Простое сохранение старого файла, обновление только его временной метки или запись маркера завершения не является публикацией. Предоставленный 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
У этого пакета другие имена стендов и три партиции вместо четырёх. Сохраните те же файлы программы. Разбор аргументов выбирает его настоящие топик и каталог, а обнаружение находит полный набор реальных партиций:
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Ожидаются шесть физических записей, четыре логических голоса и 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' "$?"
Ожидаются ненулевой статус завершения и Conflicting event_id, указывающий настоящий ID. Правило выполняется до того, как подготовленный механизм записи создаёт/заменяет результат. Способ с .pending/заменой здесь служит для публикации полного результата, но не гарантирует восстановление после аварии или долговечность данных.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
Корректный основной файл должен по-прежнему соответствовать отпечатку и сохранять исходные счётчики студии. Неуспешный конфликт записывается отдельно от успешных запусков main/check. Не редактируйте конфликтный источник и не выбирайте событие, пришедшее первым. Обратный или последовательный обход партиций также должен отклонять тот же конфликт.
При обычной опечатке в имени пакета точка входа выводит доступные имена до начала повторного чтения. Используйте vote_tool.py list и повторите запуск с правильным именем. При локальной ошибке программы исправьте файл и повторно запустите затронутый корректный пакет. После любой правки исходной программы заново запускайте оба корректных пакета, чтобы результаты относились к текущим файлам, а затем повторяйте самопроверку конфликта.
Держите группы исходных приложений отображения приостановленными: вступление, фиксация смещений, сброс и удаление запрещены. Сброс с последующим восстановлением того же конечного смещения также не относится к независимому повторному чтению. Функциональные проверки объединяют нативное получение деревом программы с настоящими полными результатами; они не доказывают причинность вычисления или устойчивость к администратору, подделывающему окружение.
Резюме
Вы получили настоящие границы каждой партиции и прочитали полные конечные журналы, несмотря на временно пустые опросы, пустые партиции и пропуски позиций. Та же нативная программа восстановила результаты для двух каталогов без накопления старых сумм, а конфликтующий ID вызвал ошибку с сохранением полезного основного снимка и исходного прогресса консьюмера.
