Introdução
O produtor de console demonstrou o modelo de eventos e chaves. Uma aplicação precisa de mais uma distinção: chamar uma API de envio pode colocar um registro na fila local do cliente antes de o broker confirmá-lo. Um programa que encerra nesse momento pode informar sucesso sem entregar seus registros.
Você usará o cliente Python confluent-kafka instalado com o broker local do Apache Kafka preparado. O pacote é uma biblioteca cliente; nenhuma conta no Confluent Cloud é necessária. Você serializará dicionários em bytes JSON/UTF-8, enviará chaves de pedidos e inspecionará callbacks de entrega e o estado final da fila. Um endereço local inacessível fornece um teste de falha com prazo limitado.
Este laboratório independente prepara um tópico com três partições e dois arquivos de entrada. As atividades posteriores do curso pressupõem conhecimentos de funções, laços, dicionários, JSON e execução de scripts em Python; a trilha de Python pode ajudar com esses pré-requisitos. A instalação do Kafka e a configuração do serviço continuam preparadas. Use o Terminal.
Transforme um dicionário em chave e valor para transmissão
Nesta etapa, você inspecionará a conversão antes de contatar o Kafka. Comece pela entrada de seis eventos:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Cada linha é um objeto JSON completo. json.loads converte esse texto em um dicionário Python. json.dumps converte o dicionário em texto JSON, e encode("utf-8") converte texto em bytes adequados à API do cliente. A chave é codificada separadamente a partir de order_id; ela não é event_id.
Salve e execute esta pequena prévia. O here-document com delimitador entre aspas preserva literalmente o código Python:
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Localize o dicionário, o texto JSON e as representações em bytes. A exibição de bytes do Python começa com b; essa notação de exibição não faz parte do valor transmitido. A primeira chave deve codificar order-4001. O JSON preserva os campos nomeados do evento, enquanto UTF-8 fornece a codificação de caracteres.
Nenhum evento foi enviado ainda. A próxima etapa fornece esses bytes a um produtor e verifica quando o broker os confirma.
Publique e confirme cada entrega real
Nesta etapa, você executará um produtor completo com tratamento de argumentos e callback de entrega fornecidos. Producer configura o cliente; bootstrap.servers fornece o endereço inicial. acks=all solicita confirmação das réplicas atualmente sincronizadas. Este ambiente com um broker tem uma réplica, portanto essa configuração não demonstra tolerância a falhas entre vários brokers. A configuração de direcionamento murmur2_random fornecida mantém o direcionamento por chave compatível com os exemplos anteriores do console Java enquanto a quantidade de partições permanecer fixa.
Observe a distinção principal no programa: produce coloca um registro na fila local. poll processa callbacks, e flush aguarda o trabalho na fila enquanto também processa callbacks.
Um callback é uma função que você fornece à biblioteca cliente para que ela a chame quando um resultado de entrega estiver disponível. Passar callback=delivery fornece a própria função; isso não a chama imediatamente. poll(0) processa callbacks disponíveis sem esperar, enquanto flush também aguarda envios pendentes. O callback registra uma posição de tópico/partição/offset confirmada pelo broker ou uma falha. Após flush, o programa verifica registros restantes, falhas e quantidade de confirmações antes de declarar sucesso. Um callback de falha é coletado e causa um resultado de processo diferente de zero; ele não é apenas impresso e ignorado silenciosamente.
A seção argparse fornecida mapeia opções de linha de comando para atributos de args: por exemplo, --input se torna args.input, e --delivery-timeout-ms se torna args.delivery_timeout_ms. Path(args.input).open(...) abre o arquivo local selecionado; o laço lê um evento JSON por linha não vazia. Você não precisa projetar um analisador de argumentos para usar este cliente.
Salve o programa completo. A estrutura do analisador e do callback é fornecida, enquanto a serialização e a chamada a produce aplicam os bytes que você acabou de inspecionar:
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Execute o produtor e salve os dois fluxos de saída. > direciona a saída padrão ao log; 2>&1 envia a saída de erro padrão ao mesmo destino, preservando também as falhas de callback e os diagnósticos de conexão.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Espere seis linhas de callback DELIVERED e Delivered records: 6. A partição e o offset reais de cada callback podem variar. Um retorno bem-sucedido de produce, sozinho, não comprovaria essas confirmações.
Inspecione o broker de forma independente com o consumidor de console já aprendido:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Localize de py1 a py6 com chaves correspondentes aos valores de order_id. A leitura independente conecta o relatório de entrega da aplicação aos registros reais do broker; nenhum número fixo de partição nem ordem global de exibição é exigido.
Altere um campo de um novo evento e reutilize o produtor
Nesta etapa, você reutilizará a mesma aplicação para um novo evento. O arquivo preparado next-event tem um event_id novo, py7, portanto ainda não foi publicado. Inspecione seu campo channel:
cat /home/labex/project/data/next-event.jsonl
Altere apenas o campo channel desse evento ainda não publicado de online para store em um arquivo de trabalho separado. O evento continua se referindo a order-4001 e segue seus eventos anteriores com sequence 3:
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Espere um novo callback bem-sucedido e sete eventos retidos na leitura independente. py7 deve ter channel store, enquanto de py1 a py6 mantêm seu conteúdo original. Os eventos relacionados de order-4001 seguem sua sequência dentro da mesma partição de destino; a ordem de exibição entre partições pode variar.
A implementação do produtor não mudou. Um caminho de entrada recebido permitiu serializar e entregar os novos dados em vez de incorporar uma resposta fixa de treinamento.
Observe uma falha com prazo limitado em um endereço inacessível
Nesta etapa, você verificará que colocar na fila não é confundido com entregar. O mesmo cliente pode criar uma fila local mesmo quando localhost:1 não tem um broker Kafka. A opção de tempo limite de entrega fornecida limita a conclusão da mensagem a 3000 milissegundos, e flush acrescenta uma pequena margem limitada. O broker real continua em localhost:9092.
Envie o arquivo de um evento para esse endereço deliberadamente incorreto. O próximo comando deve falhar.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Capture o status imediatamente, antes que outro comando o substitua. $? é o status de saída do comando anterior: zero para sucesso, diferente de zero para falha. printf o salva como uma linha de texto:
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Espere um status diferente de zero, diagnósticos de conexão, um callback DELIVERY_FAILED e Delivery incomplete em vez de uma contagem final de sucesso. O programa verificou falhas dos callbacks mesmo quando a fila terminou de esvaziar após as mensagens falharem.
O endereço com falha não deve ter acrescentado um segundo py7 ao tópico real. Os sete eventos anteriores entregues com sucesso continuam disponíveis. Corrigir um endereço e tentar novamente um evento de negócio é uma decisão separada de presumir que um registro na fila foi confirmado; a falha deliberada deste laboratório não enviou nada ao broker real. A seguir, você decodificará JSON recebido, salvará os dados e confirmará o progresso de leitura em um consumidor Python.
Resumo
Você serializou dicionários e chaves de pedidos em bytes UTF-8, publicou por um cliente Python real e conferiu os callbacks bem-sucedidos com as posições reais no broker. Um novo evento alterado reutilizou o mesmo programa, enquanto um endereço inacessível produziu uma falha com prazo limitado e status diferente de zero em vez de um falso sucesso. A seguir, um consumidor Python tornará explícitos a decodificação, a saída em arquivo e o comportamento de gravar antes de confirmar.
