Data Pipelines, Streaming & dbt

Kafka, Airflow, and the Transformation Layer

Data doesn't magically appear in a dashboard. It flows through a carefully orchestrated **Pipeline**: ingested by **Apache Kafka**, transformed by **dbt** (data build tool), and orchestrated by **Apache Airflow**. For **Walmart**, this is how millions of IoT sensor readings become a clean, queryable inventory report. For **Visa**, it's how raw transaction streams become real-time fraud scores.

In this chapter, we explore the ETL vs. ELT debate, build a streaming pipeline with Kafka, create modular SQL transformations with dbt, and orchestrate the entire flow with Airflow DAGs.

Kafka: Real-Time Streaming Ingestion

**Apache Kafka** is a distributed event streaming platform. For **Walmart**, Kafka ingests millions of IoT sensor pulses per second from trucks, warehouses, and stores. For **Netflix**, it captures every click, hover, and play event in real-time. Kafka acts as the "nervous system" of the modern data stack.

PythonRuns entirely in your browser — nothing is sent to a server.

# Walmart IoT: Kafka Streaming Simulation
class KafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.buffer = []

    def send(self, event):
        self.buffer.append(event)
        print("KAFKA PRODUCE -> Topic '{0}': {1}".format(self.topic, event))

class KafkaConsumer:
    def __init__(self, topic):
        self.topic = topic

    def consume(self, producer_buffer):
        print("\nKAFKA CONSUME <- Topic '{0}':".format(self.topic))
        for event in producer_buffer:
            temp = event.get("temp_c", 0)
            status = "ALERT" if temp > 7.0 else "OK"
            print(" - Truck {0}: {1}C -> {2}".format(event.get("truck_id"), temp, status))

# Scenario: Walmart cold-chain monitoring
producer = KafkaProducer(topic="iot-cold-chain")
producer.send({"truck_id": "TX-42", "temp_c": 4.2})
producer.send({"truck_id": "TX-43", "temp_c": 8.9})
producer.send({"truck_id": "TX-44", "temp_c": 3.8})

consumer = KafkaConsumer(topic="iot-cold-chain")
consumer.consume(producer.buffer)

# Walmart IoT: Kafka Streaming Simulation
class KafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.buffer = []

    def send(self, event):
        self.buffer.append(event)
        print("KAFKA PRODUCE -> Topic '{0}': {1}".format(self.topic, event))

class KafkaConsumer:
    def __init__(self, topic):
        self.topic = topic

    def consume(self, producer_buffer):
        print("\nKAFKA CONSUME <- Topic '{0}':".format(self.topic))
        for event in producer_buffer:
            temp = event.get("temp_c", 0)
            status = "ALERT" if temp > 7.0 else "OK"
            print(" - Truck {0}: {1}C -> {2}".format(event.get("truck_id"), temp, status))

# Scenario: Walmart cold-chain monitoring
producer = KafkaProducer(topic="iot-cold-chain")
producer.send({"truck_id": "TX-42", "temp_c": 4.2})
producer.send({"truck_id": "TX-43", "temp_c": 8.9})
producer.send({"truck_id": "TX-44", "temp_c": 3.8})

consumer = KafkaConsumer(topic="iot-cold-chain")
consumer.consume(producer.buffer)

dbt: The Transformation Layer

**dbt (data build tool)** transforms raw data inside your warehouse using modular, version-controlled SQL. For **Visa**, dbt models take raw transaction logs and create clean "marts" (business-ready tables). The power of dbt is its **Lineage Graph**: you can see exactly how every table was derived, making auditing transparent.

PythonRuns entirely in your browser — nothing is sent to a server.

# Visa Analytics: dbt Model Simulation
class DbtModel:
    def __init__(self, model_name, source_data):
        self.name = model_name
        self.source = source_data
        self.output = []

    def transform(self):
        print("DBT MODEL: '{0}'".format(self.name))
        print(" - Source: {0} raw records".format(len(self.source)))
        
        # Simulating a dbt SQL transformation
        self.output = [
            {"merchant": r["merchant"], "total": r["amount"], "is_high_value": r["amount"] > 500}
            for r in self.source
        ]
        
        for row in self.output:
            flag = "HIGH-VALUE" if row["is_high_value"] else "standard"
            print(" - {0}: 1 [{2}]".format(row["merchant"], row["total"], flag))
        
        print("\nDBT LINEAGE: raw_transactions -> stg_transactions -> mart_merchant_summary")

# Scenario: Visa's daily merchant aggregation
raw_tx = [
    {"merchant": "Amazon", "amount": 1200},
    {"merchant": "Netflix", "amount": 15},
    {"merchant": "Amazon", "amount": 800}
]
model = DbtModel("mart_merchant_summary", raw_tx)
model.transform()

# Visa Analytics: dbt Model Simulation
class DbtModel:
    def __init__(self, model_name, source_data):
        self.name = model_name
        self.source = source_data
        self.output = []

    def transform(self):
        print("DBT MODEL: '{0}'".format(self.name))
        print(" - Source: {0} raw records".format(len(self.source)))
        
        # Simulating a dbt SQL transformation
        self.output = [
            {"merchant": r["merchant"], "total": r["amount"], "is_high_value": r["amount"] > 500}
            for r in self.source
        ]
        
        for row in self.output:
            flag = "HIGH-VALUE" if row["is_high_value"] else "standard"
            print(" - {0}: 1 [{2}]".format(row["merchant"], row["total"], flag))
        
        print("\nDBT LINEAGE: raw_transactions -> stg_transactions -> mart_merchant_summary")

# Scenario: Visa's daily merchant aggregation
raw_tx = [
    {"merchant": "Amazon", "amount": 1200},
    {"merchant": "Netflix", "amount": 15},
    {"merchant": "Amazon", "amount": 800}
]
model = DbtModel("mart_merchant_summary", raw_tx)
model.transform()

Airflow: Orchestrating the Pipeline

**Apache Airflow** is the scheduler that ties everything together. It uses **DAGs (Directed Acyclic Graphs)** to define task dependencies: "First ingest from Kafka, then run dbt transformations, then trigger the ML model refresh." For **Walmart**, Airflow ensures the morning inventory report is always built from the freshest data.

PythonRuns entirely in your browser — nothing is sent to a server.

# Walmart Pipeline: Airflow DAG Simulation
class AirflowDAG:
    def __init__(self, dag_name):
        self.name = dag_name
        self.tasks = []

    def add_task(self, task_name, depends_on=None):
        self.tasks.append({"name": task_name, "depends_on": depends_on})

    def execute(self):
        print("AIRFLOW DAG: '{0}'".format(self.name))
        print("Execution Order:")
        for i, task in enumerate(self.tasks):
            dep = task["depends_on"] or "None"
            print(" [{0}] {1} (depends_on: {2})".format(i + 1, task["name"], dep))
        
        print("\nRunning tasks in dependency order...")
        for task in self.tasks:
            print(" RUNNING: {0}... COMPLETE".format(task["name"]))
        print("\nDAG '{0}' finished successfully.".format(self.name))

# Scenario: Walmart's daily inventory refresh pipeline
dag = AirflowDAG("daily_inventory_refresh")
dag.add_task("ingest_kafka_iot")
dag.add_task("run_dbt_cleaning", depends_on="ingest_kafka_iot")
dag.add_task("run_dbt_aggregation", depends_on="run_dbt_cleaning")
dag.add_task("refresh_ml_demand_model", depends_on="run_dbt_aggregation")
dag.add_task("update_dashboard", depends_on="refresh_ml_demand_model")
dag.execute()

# Walmart Pipeline: Airflow DAG Simulation
class AirflowDAG:
    def __init__(self, dag_name):
        self.name = dag_name
        self.tasks = []

    def add_task(self, task_name, depends_on=None):
        self.tasks.append({"name": task_name, "depends_on": depends_on})

    def execute(self):
        print("AIRFLOW DAG: '{0}'".format(self.name))
        print("Execution Order:")
        for i, task in enumerate(self.tasks):
            dep = task["depends_on"] or "None"
            print(" [{0}] {1} (depends_on: {2})".format(i + 1, task["name"], dep))
        
        print("\nRunning tasks in dependency order...")
        for task in self.tasks:
            print(" RUNNING: {0}... COMPLETE".format(task["name"]))
        print("\nDAG '{0}' finished successfully.".format(self.name))

# Scenario: Walmart's daily inventory refresh pipeline
dag = AirflowDAG("daily_inventory_refresh")
dag.add_task("ingest_kafka_iot")
dag.add_task("run_dbt_cleaning", depends_on="ingest_kafka_iot")
dag.add_task("run_dbt_aggregation", depends_on="run_dbt_cleaning")
dag.add_task("refresh_ml_demand_model", depends_on="run_dbt_aggregation")
dag.add_task("update_dashboard", depends_on="refresh_ml_demand_model")
dag.execute()

Practice Questions

Question 1

Why is Kafka described as the 'nervous system' of modern data architecture?

Because it stores data permanently
Because it acts as the real-time event backbone, ingesting and distributing millions of events per second to downstream consumers
Because it only works with Netflix
Because it replaces the need for a database

Question 2

What is the primary benefit of dbt's lineage graph?

It makes SQL queries colorful
It provides complete transparency into how every table was derived, enabling auditing and impact analysis
It speeds up query execution
It replaces the need for Kafka