Introdução
Uma pequena galeria permite que os visitantes votem no estande favorito. O remetente pode reenviar um evento, portanto um novo registro Kafka nem sempre significa mais um voto. Primeiro você observará um exemplo normal com dois votos, depois um reenvio real em outra posição armazenada e, por fim, reconstruirá a contagem da galeria.
Você já deve conhecer caminhos de terminal, funções/laços/dicionários Python, JSON e os fundamentos de chaves, partições, offsets, grupos de consumidores e cliente Python do Kafka apresentados em Apache Kafka for Beginners. Este laboratório acrescenta a identidade do evento de negócio. Um leitor nativo completo está preparado para que você se concentre na regra de contagem; o próximo laboratório guiado constrói esse leitor. Use Terminal e o ambiente bigdata-python instalado. A estimativa de 40–50 minutos do autor inclui leitura e erros comuns, não uma medição com iniciantes.
O trabalho envolve apenas um lote finito já entregue. Não há sistema de elegibilidade de votantes, fluxo contínuo, garantia de recuperação após falhas abruptas ou processamento exatamente uma vez de ponta a ponta. Cada laboratório começa de forma independente.
Observe um reenvio em outra posição armazenada
Nesta etapa, você comparará dois votos normais com um evento reenviado. Comece em Terminal. A área de trabalho contém ferramentas e catálogos preparados. Edite apenas reducer.py, replay.py e auxiliares Python opcionais em helpers/; coloque arquivos derivados em output/. Preserve as ferramentas/dados fornecidos, os tópicos de origem congelados e os grupos dos aplicativos de exibição pausados. O tópico separado de demonstração permite uma única ação: o reenvio abaixo.
cd /home/labex/project
-B impede que esses comandos Python criem arquivos de cache de bytecode. A ferramenta somente de leitura inspect mostra os envelopes reais dos registros do broker: tópico, partição, offset, chave e valor decodificado. A contagem final é o número de registros de aplicativo armazenados. Inspecione a pequena demonstração:
bigdata-python -B vote_tool.py inspect demo
Você deve ver dois registros. Seus valores seguem este esquema: event_id, booth e choice são strings não vazias; choice é exatamente favourite; booth deve aparecer no catálogo selecionado. Nenhum campo extra pertence ao esquema de voto deste curso. A chave é o nome do estande, usado no roteamento. event_id identifica um voto de negócio. Aqui, demo-1 vota em oak e demo-2 em fern: dois registros e dois votos lógicos.
Antes de enviar, preveja os dois resultados: quantos registros físicos serão armazenados e quais serão as contagens dos estandes se demo-1 continuar sendo um único voto de negócio? Depois teste essa previsão.
O comando de reenvio envia o mesmo valor completo de demo-1 ao broker real e aguarda a confirmação da entrega. Antes de enviar, ele verifica a demonstração existente; por isso, executá-lo novamente depois do reenvio esperado não acrescenta um quarto registro:
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
Agora há três registros. demo-1 aparece em dois offsets diferentes com conteúdo de valor idêntico. Essas posições identificam dois registros armazenados, mas o ID de evento repetido continua identificando um só voto. Os totais corretos permanecem oak=1, fern=1, reed=0. Ler novamente um registro existente e armazenar outro reenvio são situações diferentes; ambas precisam de uma regra de identidade de negócio ao construir as contagens.
O lote congelado do curso usa posicionamento explícito nas partições, portanto chaves de estande iguais podem aparecer em partições diferentes. Leia os metadados reais ao comparar os registros.
Dentro do lote selecionado do curso, o ID de evento é global em todas as partições, e não apenas em uma partição ou estande. Dois IDs diferentes contam como dois votos mesmo quando todos os outros campos são iguais. As regras de negócio fornecidas aceitam esses eventos; elas não decidem quantos votos uma pessoa real pode dar.
Se uma ferramenta indicar um lote desconhecido, liste os nomes preparados e use demo nesta observação:
bigdata-python -B vote_tool.py list
Um erro de digitação não reiniciou o broker. Corrija o argumento e repita a inspeção somente de leitura. Não reenvie eventos arbitrários para corrigir uma contagem exibida.
Preserve as identidades lógicas e recuse conteúdo conflitante
Nesta etapa, você implementará a regra de negócio dentro da função Python preparada. Continue em Terminal, em /home/labex/project. Leia o arquivo atual:
cat reducer.py
Ele inicializa as contagens do catálogo e depois incrementa para cada valor de entrada. Preveja o que faria com os três registros da demonstração: o segundo demo-1 armazenado acrescentaria outro voto. A regra correta deve contar esse ID uma vez e manter os IDs distintos separados.
O verificador preparado chama uma cópia exata da sua função em exemplos descartáveis, incluindo valores reordenados e nomes de estandes diferentes. Ele não altera o log ativo do Kafka. Execute a função ingênua atual; este comando deve falhar com uma mensagem sobre a regra:
bigdata-python -B check_rules.py
O shell guarda o status de saída do comando imediatamente anterior em $?: zero significa sucesso e um valor diferente de zero significa falha. Mostre-o imediatamente, antes que outro comando o substitua:
printf '%s\n' "$?"
summarize(values, booths) deve retornar um dicionário contendo votes e counts para entradas válidas e lançar uma exceção comum para entradas inválidas. Mantenha o módulo chamável; não encerre o interpretador Python a partir dele. O verificador precisa receber os valores realmente retornados, não apenas um status de saída do interpretador.
Edite a função existente no nano
Abra o arquivo editável no editor de texto do Terminal:
nano reducer.py
A barra superior identifica o nome do arquivo. Use as setas para se mover e Backspace/Delete para corrigir caracteres. O Python usa indentação para agrupar instruções: use quatro espaços dentro da função, oito dentro do laço e mais quatro para cada condição aninhada. Mantenha a linha existente def summarize(values, booths):. Remova o comentário antigo sobre contar registros intencionalmente ao substituir essa lógica.
Os blocos Python curtos a seguir são edições conectadas dentro desta única função, não comandos para executar no shell nem programas separados. Complete todas as edições antes de executar o verificador. Preserve as partes válidas do código inicial em vez de substituir o arquivo inteiro.
Primeiro, mantenha a inicialização existente das contagens com zero e acrescente seen logo abaixo, antes do laço. Um dicionário indexado por event_id guardará o valor completo de cada voto de negócio aceito:
counts = {booth: 0 for booth in booths}
seen = {}
Em seguida, mantenha a única linha existente for value in values: e substitua seu incremento original pela validação abaixo. O bloco mostra essa linha for existente como contexto; não acrescente um segundo laço. set(value) reúne nomes de campos; isinstance verifica tipos; any(...) pergunta se pelo menos um valor de campo é inválido. A primeira condição exige um dicionário com exatamente três campos de string não vazia e a escolha aprovada. Verifique o catálogo antes de tentar incrementar um estande. Mantenha quatro espaços na linha for e oito na validação:
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"])
Agora acrescente o tratamento de identidade imediatamente após a verificação do catálogo, ainda dentro do laço. Para um ID conhecido, compare o valor decodificado completo. Conteúdo diferente deve lançar um erro que cite esse ID; escolher o primeiro ou o último valor faria a resposta depender da ordem de percurso. Um reenvio idêntico chega a continue, que passa à próxima entrada sem outro incremento. Um ID visto pela primeira vez é salvo e contado:
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) conserva o valor aceito como dicionário. Espaços e ordem dos campos JSON não constituem identidade de negócio. A chave do dicionário é global nas partições do lote selecionado; não é um nome de estande nem um par partição/offset. IDs diferentes com o mesmo estande e escolha devem, cada um, chegar ao incremento de primeiro ID.
Por fim, substitua o retorno antigo que usava todos os valores de entrada. Retorne os valores únicos aceitos e todas as contagens do catálogo. Esta linha tem quatro espaços, fora do laço, mas dentro da função:
return {"votes": list(seen.values()), "counts": counts}
Salve e saia. Nos rótulos inferiores do nano, ^ significa manter Ctrl pressionado: pressione Ctrl+O, depois Enter para confirmar reducer.py; pressione Ctrl+X para voltar ao shell. Essas teclas salvam o arquivo atual em vez de executar Python. Leia o que você realmente salvou:
cat reducer.py
Confira se seen é inicializado uma vez por chamada da função, se as verificações de esquema/catálogo acontecem antes da contagem, se apenas um primeiro ID incrementa e se o retorno está fora do laço. Depois teste imediatamente a função completa:
bigdata-python -B check_rules.py
Espere Rules pass para identidades, IDs distintos, conflito, esquema, estandes com zero votos e entrada reordenada. Um dicionário indexado somente pelo estande juntaria eventos de voto distintos; um indexado por partição/offset manteria todos os reenvios. Seu dicionário representa a identidade de negócio.
Para um erro de indentação, sintaxe ou nome, leia o nome de arquivo e a linha informados, execute nano reducer.py novamente e use as setas para corrigir a parte afetada. Salve com Ctrl+O, Enter, saia com Ctrl+X, examine com cat reducer.py e repita a verificação da regra. Preserve as partes já corretas; não é preciso reescrever a função inteira nem reiniciar a VM. Um caminho incorreto pode ser localizado com estes comandos somente de leitura:
pwd
ls reducer.py replay.py
Volte com cd /home/labex/project, se necessário. A próxima etapa usa o leitor nativo preparado para aplicar sua regra a registros reais do Kafka.
Publique e leia um snapshot do log real da galeria
Nesta etapa, você aplicará sua função ao lote real congelado da galeria. Continue em Terminal, em /home/labex/project. Liste a origem e a saída escolhidas pelo lote main e depois examine seu catálogo:
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main corresponde a gallery-votes. Ele tem várias partições, IDs repetidos em posições armazenadas realmente diferentes, IDs distintos com o restante do conteúdo igual e uma partição vazia. replay.py é o leitor nativo preparado; vote_support.py fornece análise de argumentos e escrita de snapshots. O leitor passa o valor de cada envelope real à sua função summarize. Ele não obtém valores de um arquivo local de entrada de votos.
Execute o ponto de entrada público abaixo. Ele primeiro verifica a função de identidade atual em exemplos descartáveis, depois executa seu programa atual e seus processos filhos auxiliares, conserva evidências de leitura nativa dessa árvore de processos e salva o resultado bem-sucedido no caminho de saída indicado:
bigdata-python -B run_replay.py main
Espere uma publicação com records=8, logical_votes=5. O ponto de entrada registra o recebimento real de dados do broker por essa árvore de programas, em vez de confiar em um consumidor alheio com o mesmo rótulo de cliente. Ele não demonstra que o cálculo interno realmente usou esses dados nem oferece proteção contra falsificação por um administrador. O resultado de negócio final é verificado separadamente contra a origem real completa.
Cada execução bem-sucedida deve publicar realmente o resultado durante essa invocação: escrever dados diretamente no arquivo ou escrever um arquivo temporário e renomeá-lo para o resultado. Um processo filho auxiliar pode realizar a escrita. Publicar novamente bytes idênticos é válido; apenas ler o Kafka e manter um arquivo antigo não é uma nova publicação.
Leia o arquivo de resultado real. Este leitor nunca busca, completa ou repara silenciosamente resultados ausentes:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Espere cinco IDs de votos lógicos e estas contagens por estande:
Booth Votes
fern 2
oak 3
reed 0
O snapshot também preserva os oito envelopes físicos, permitindo distinguir seus registros de origem dos cinco votos lógicos. reed continua presente com zero. Com o mecanismo de escrita preparado, erros de laço ou validação anteriores à publicação deixam o snapshot anterior disponível. O ponto de entrada não desfaz automaticamente outros métodos de escrita. Após qualquer edição do programa, execute novamente o ponto de entrada público antes de verificar a conclusão.
O aplicativo de exibição original está pausado com o progresso de seu grupo de consumidores salvo. Você pode descrever esse grupo, mas não pode entrar nele, redefini-lo/excluí-lo nem confirmar offsets nele. Não edite/exclua o tópico de origem nem acrescente partições. Reconstrua apenas sua própria saída derivada. A verificação observa a identidade original do tópico, a topologia completa de partições/réplicas, os registros reais e o progresso salvo do aplicativo; ela não promete preservar qualquer configuração arbitrária do broker.
Agora você tem um snapshot útil dos estandes, não apenas um histórico de entregas. O próximo laboratório explica e constrói o leitor nativo independente completo, incluindo por que um único polling vazio não significa que o lote terminou.
Resumo
Você observou um reenvio real armazenado, separou a identidade global de negócio das posições de registros Kafka e recusou conteúdo conflitante com o mesmo ID. Seu snapshot real da galeria preserva a origem física completa enquanto apresenta os votos lógicos e todos os estandes, inclusive os que têm zero votos. A origem e o grupo do aplicativo de exibição pausado continuam disponíveis.
