Introdução
Um estúdio preservou o histórico de votos enquanto seu aplicativo de exibição original está pausado. Você precisa de uma nova contagem sem retroceder esse aplicativo nem alterar o log. Um polling pode não retornar mensagens por um breve período, mesmo que um consumidor novo ainda não tenha recebido o lote congelado.
Você construirá o laço de releitura nativa, capturará separadamente a posição final de cada partição e publicará somente depois de ler o lote finito completo. A função de identidade lógica do laboratório anterior já está disponível nesta VM independente. São necessários os fundamentos de Python e Kafka usados lá. Use Terminal. A estimativa de 50–60 minutos do autor inclui leitura e erros comuns; não é um teste cronometrado com iniciantes.
Esta é uma reconstrução de lote congelado. Não há novas publicações nesses tópicos durante o exercício, nenhuma implementação de transações a construir e nenhuma promessa de recuperação confiável após falhas abruptas ou de processamento exatamente uma vez de ponta a ponta.
Capture o limite de cada partição e leia até ele
Nesta etapa, você implementará uma leitura nativa independente completa dentro de um programa externo preparado. Comece em Terminal. Edite replay.py e, se necessário, reducer.py ou auxiliares Python em helpers/; escreva arquivos derivados em output/. Preserve as ferramentas/dados fornecidos, os tópicos originais e os grupos dos aplicativos de exibição pausados:
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
A descrição somente de leitura mostra os IDs reais das partições de studio-votes, o início retido low, o fim congelado atual e as réplicas. Uma partição é um log ordenado, não uma VM separada. O fim é a próxima posição depois do lote retido, não um registro a contar. Preveja quais partições precisam de registros: uma partição vazia tem low == high e já está completa.
Um intervalo de offsets não é uma contagem de registros. O Kafka pode pular posições internas/de controle ou removidas. Esta origem inclui uma posição nativa pulada; não há voto ausente a inventar. Inspecione os envelopes reais e compare suas posições reais com a descrição:
bigdata-python -B vote_tool.py inspect main
Antes de reconstruir, leia as posições salvas pelo aplicativo de exibição pausado com a ferramenta de grupos de consumidores do curso prévio:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Guarde cada CURRENT-OFFSET para comparar mais tarde. Não haver membros ativos é normal: o aplicativo está pausado, mas tem progresso salvo. Aqui, LAG é uma diferença entre posições do log, não uma contagem de votos lógicos. Seu consumidor independente não deve avançar nem retroceder o progresso desse grupo.
Leia a estrutura externa do programa preparado
cat replay.py
As importações, arguments(), a configuração do cliente, a comunicação de erros e finally/fechamento já estão fornecidos. O núcleo lança deliberadamente NotImplementedError; ele não lê partições nem publica uma resposta de exemplo. A função de identidade correta do laboratório anterior está preparada separadamente em reducer.py.
O cliente usa o endereço bootstrap.servers selecionado e um group.id independente. allow.auto.create.topics é falso. O commit automático e o armazenamento automático de offsets estão desativados. assign fornecerá posições iniciais explícitas sem alocação dinâmica de grupo; este exemplo não faz chamada explícita de commit. Uma implementação equivalente com grupo independente pode salvar seu próprio progresso, mas nunca deve entrar em um grupo de exibição original, fazer commit nele, redefini-lo ou excluí-lo.
O vote_support.py preparado fornece arguments() para tópico/catálogo/saída/broker/grupo e finish(...) para escrever o resultado completo. Ele não descobre partições nem busca votos. finish chama a regra de identidade antes de escrever/substituir o resultado. finally fecha o consumidor após sucesso ou falha; mantenha essa estrutura externa.
Implemente o núcleo nativo em partes curtas conectadas
Use o fluxo de edição do laboratório anterior:
nano replay.py
Mantenha as importações, a análise de argumentos, a configuração de Consumer e as seções try:, except e finally. Dentro de try:, remova apenas os dois comentários explicativos e o raise NotImplementedError(...) do núcleo incompleto. Use as setas para posicionar o cursor e Backspace/Delete para editar; Ctrl+K pode recortar a linha atual inteira, então use-o apenas quando estiver em uma linha que deseja substituir.
Insira as partes conectadas a seguir nesse espaço. Elas formam um único núcleo: não execute blocos isolados no shell nem salve/execute um laço pela metade. Complete todas as partes, mantenha a estrutura externa de erro/fechamento fornecida e depois salve e execute. Quatro espaços colocam uma instrução dentro de try:; outros quatro a colocam dentro de um laço ou condição. A indentação relativa mostrada faz parte do significado do Python.
Primeiro descubra o conjunto real completo de partições. Chamar list_topics sem solicitar um tópico evita pedir metadados de um nome desconhecido. Confira a disponibilidade antes de usar a entrada selecionada. Ordenar os IDs descobertos dá a este exemplo uma lista estável de atribuição; não cria uma ordem global de eventos:
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)
Para cada partição descoberta, solicite get_watermark_offsets(..., timeout=10) uma vez. TopicPartition identifica o tópico e a partição dessa solicitação. Guarde tanto o low retido quanto o high congelado e inicialize a próxima posição de diagnóstico em low. Não atualize high dentro do laço de polling: perseguir um fim crescente deixaria de ser uma recontagem finita. Este exercício proíbe acrescentar dados à origem congelada ou alterá-la:
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
Construa pending apenas com partições que tenham posições retidas a ler. assign recebe uma lista de offsets iniciais explícitos, incluindo os inícios low das partições vazias. Inicialize uma nova lista de registros físicos a cada execução. time.monotonic() mede o tempo decorrido independentemente de ajustes no relógio do sistema; o prazo limita uma leitura incompleta:
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
Agora comece o laço. poll(0.2) pode retornar uma mensagem, uma mensagem com erro/evento ou None quando a espera expira. Preveja por que None não pode significar conclusão: o cliente pode simplesmente ainda não ter recebido um registro congelado. Continue aguardando dentro do prazo total. Eventos EOF de partição estão habilitados na configuração; _PARTITION_EOF é o evento de fim esperado, enquanto outros erros fazem a execução falhar. Isso permite o avanço nativo além de uma posição interna final:
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()))
Continue o mesmo bloco if message is not None com o elif abaixo, alinhado com if message.error(): em doze espaços. Somente uma mensagem normal em uma partição pendente, abaixo de seu high fixo, torna-se um envelope. Preserve também os reenvios físicos; summarize trata a identidade lógica depois. Decodifique os bytes UTF-8 da chave/valor; em seguida, json.loads transforma o valor em um dicionário:
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")),
})
Depois de tratar esse polling, consulte consumer.position(...) para cada partição restante. Essa é a próxima posição do consumidor, distinta do progresso confirmado de qualquer grupo. Compare cada uma com seu high capturado, não com uma contagem presumida de registros. Remova de pending somente as partições concluídas. A linha for position tem oito espaços, dentro de while, mas fora da condição da mensagem, de modo que mesmo um polling ocioso pode ser seguido por uma verificação de posição. Por fim, chame finish com quatro espaços, depois da conclusão do laço:
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)
Agora cada partição tem um limite fixo. As vazias não precisaram de voto; posições puladas não exigiram registros inventados; None não interrompeu a leitura. Um snapshot completo só é publicado depois que todas as próximas posições pendentes atingem seus finais high.
Salve com Ctrl+O, Enter para confirmar replay.py e depois Ctrl+X. Examine o programa salvo:
cat replay.py
Confira se todas as seis partes estão dentro do try fornecido, se o tratamento de envelopes permanece no ramo de mensagens normais, se a verificação de posição fica dentro do laço e se finish está fora dele. Execute imediatamente no lote nativo selecionado:
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
Espere oito registros físicos, cinco votos lógicos e cedar=3, maple=2, willow=0. O ponto de entrada público primeiro verifica a função de identidade em exemplos descartáveis e depois registra os dados do broker recebidos pela árvore nativa real do programa/auxiliares. Uma comparação de negócio completa separada verifica o resultado real. Os limites/posições informados pelo snapshot são diagnósticos, não prova independente do processamento nativo.
Cada invocação bem-sucedida deve realmente escrever seu resultado, diretamente ou por um arquivo temporário escrito e renomeado para a saída selecionada. Bytes idênticos em repetições são válidos, e processos filhos auxiliares podem escrever. Apenas manter ou atualizar a marca de tempo de um arquivo antigo, ou escrever um marcador de conclusão, não é publicação. O finish fornecido usa o método ensinado de arquivo temporário/substituição. Com esse mecanismo preparado, erros de laço ou validação antes da publicação deixam o snapshot anterior disponível; o ponto de entrada não desfaz escritas arbitrárias no próprio arquivo.
Para um erro de sintaxe/nome/indentação, use a linha informada para reabrir nano replay.py, corrigir essa parte e preservar as partes que funcionam. Salve/saia, examine com cat replay.py e execute novamente o ponto de entrada público. Para um tempo limite de releitura incompleta, examine o argumento do tópico selecionado e o tratamento de pending, EOF e posição; isso não é motivo para retroceder o grupo original. Use pwd e ls replay.py reducer.py para localizar um caminho digitado incorretamente. Um erro local de programa não exige executar o setup novamente nem alterar os registros de origem.
Reconstrua do zero e use outro lote
Nesta etapa, você provará que outra execução reconstrói o mesmo resultado finito e depois usará um catálogo e uma distribuição de partições diferentes. Continue em Terminal, em /home/labex/project.
O mecanismo de escrita preparado ordena envelopes físicos por partição/offset e votos lógicos por ID antes de salvar JSON. Isso torna estável o mesmo snapshot completo, sem afirmar uma ordem global de eventos. sha256sum calcula uma impressão digital do conteúdo. > salva essa impressão no diretório de saída; depois, -c compara o arquivo real indicado com a impressão salva:
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
Espere output/main-snapshot.json: OK. O programa começa com records, seen e contagens vazias a cada invocação. Ele substitui um snapshot completo em vez de somar novamente os votos históricos aos totais antigos. O consumo não apagou esses registros nem retrocedeu o aplicativo de exibição pausado.
Leia as posições do aplicativo original novamente após repetir a execução:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Todo CURRENT-OFFSET deve corresponder à tabela anterior do aplicativo pausado. Sua reconstrução independente não confirmou offsets nesse grupo nem o retrocedeu, embora tenha lido os registros retidos. A mensagem de ausência de membros ativos continua normal.
Liste o lote check e examine seu próprio catálogo de estandes:
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
Esse lote tem outros nomes de estandes e três partições em vez de quatro. Mantenha os mesmos arquivos de programa. O analisador escolhe seu tópico e catálogo reais, enquanto a descoberta encontra seu conjunto completo de partições reais:
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Espere seis registros físicos, quatro votos lógicos e amber=2, violet=2, moss=0. Fixar o primeiro catálogo no código ou parar na contagem de registros do primeiro tópico não descreveria corretamente esse lote.
Volte ao primeiro arquivo real:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Ele ainda contém o resultado do estúdio. O leitor abre diretamente o arquivo solicitado e não complementa nada do Kafka. Se informar um arquivo ausente, compare o nome com vote_tool.py list e corrija o caminho. Reconstrua o lote correspondente apenas quando seu resultado realmente precisar ser reconstruído.
Recuse um conflito preservando o snapshot válido
Nesta etapa, você observará um erro de dados sem substituir um resultado útil. Continue em Terminal, em /home/labex/project. O tópico conflict listado contém deliberadamente um ID com conteúdos de estande válidos diferentes. Seu destino de saída é o snapshot principal existente. Nenhum valor é um vencedor aprovado: essa execução não deve publicar o resultado como válido.
A impressão digital da etapa anterior descreve o arquivo principal válido. Execute o mesmo programa no lote conflitante. O próximo comando deve falhar:
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
Espere um status de saída diferente de zero e Conflicting event_id identificando o ID real. A regra é executada antes de o mecanismo de escrita preparado criar/substituir o resultado. O método .pending/substituição é aqui uma técnica de publicação de resultado completo, não uma garantia de recuperação após falhas abruptas nem de durabilidade.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
O arquivo principal válido deve continuar correspondendo à impressão, com as contagens originais do estúdio. O conflito que falhou é registrado separadamente das execuções bem-sucedidas de main/check. Não edite a origem do conflito nem escolha o evento que chegou primeiro. Um percurso inverso ou sequencial das partições também deve rejeitar o mesmo conflito.
Para um nome de lote digitado incorretamente, o ponto de entrada mostra os nomes disponíveis antes de começar a releitura. Consulte vote_tool.py list e execute novamente com o nome correto. Para um erro local do programa, corrija o arquivo e execute novamente o lote válido afetado. Após qualquer edição do código-fonte, execute de novo ambos os lotes válidos para que os resultados pertençam aos arquivos de programa atuais; depois repita a autoverificação do conflito.
Mantenha pausados os grupos dos aplicativos de exibição originais: não é permitido entrar, fazer commit, redefinir ou excluir. Redefinir um grupo e restaurá-lo ao mesmo offset final também fica fora da releitura independente. As verificações funcionais combinam recebimento nativo pela árvore do programa e saídas reais completas; não provam causalidade do cálculo nem resistência a um administrador que falsifique o ambiente.
Resumo
Você capturou limites reais por partição e leu os logs finitos completos apesar de pollings temporariamente ociosos, partições vazias e posições puladas. O mesmo programa nativo reconstruiu resultados para dois catálogos sem acumular totais antigos, e um ID conflitante fez a execução falhar preservando o snapshot principal útil e o progresso original do consumidor.
