Back
Data Engineering•2023

JBS Data Pipeline

PythonApache KafkaPandasSQL
JBS Data Pipeline

Overview

A high-throughput data pipeline architecture designed to ingest and process production metrics in real-time. Built on Apache Kafka, the system decouples data production from analysis, preventing data loss during high load peaks. It feeds a Data Lake that drives executive dashboards, providing granular monitoring of operational efficiency.

Links

Implementation

// production excerpt
pipeline/consumer.py
# Decouple production from analysis; survive load peaks without losing events.
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "metrics-ingest",
    "enable.auto.commit": False,             # commit only AFTER a durable write
    "security.protocol": "SASL_SSL",         # encrypted + authenticated transport
    "sasl.mechanism": "SCRAM-SHA-512",
    "ssl.ca.location": "/etc/kafka/ca.pem",
})
consumer.subscribe(["production.metrics"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        log.error("kafka.error", err=str(msg.error()))
        continue

    event = decode(msg.value())
    # Idempotent upsert keyed on event id — reprocessing never double-counts.
    lake.upsert(key=event.id, payload=event, partition=msg.partition())
    consumer.commit(msg, asynchronous=False)  # at-least-once, exactly-once effect

Encrypted + authenticated transport, manual commits, idempotent upserts.