Kafka Integration
Batch Orchestration Meeting a Streaming World
Airflow is not a streaming platform - that's been said plainly since the first module of this course. Kafka usually enters an Airflow pipeline at the edges: producing a message when a batch job finishes, or consuming a bounded batch of messages as one discrete task, never as a continuously-running stream inside a task itself.
Same honest treatment as EMR/Redshift/Glue/KubernetesPodOperator: this would need a real Kafka broker running alongside this sandbox, which isn't set up here. Code-only, no live run.
Producing: Publishing an Event
from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator
def get_order_completed_event(**context):
return [("order_id", str(context["ds"]).encode("utf-8"))]
publish_completion_event = ProduceToTopicOperator(
task_id="publish_order_completed",
kafka_config_id="kafka_default",
topic="orders.completed",
producer_function=get_order_completed_event,
)
This is the same idea as the SnsPublishOperator covered in the AWS operators section — "pipeline finished, tell the rest of the architecture" — just for teams standardized on Kafka instead of SNS.
Consuming: Waiting For and Reading Messages
from airflow.providers.apache.kafka.sensors.kafka import AwaitMessageSensor
def process_message(message):
key = message.key().decode("utf-8") if message.key() else None
return key
wait_for_upstream_event = AwaitMessageSensor(
task_id="wait_for_upstream_completed_event",
kafka_config_id="kafka_default",
topics=["upstream.completed"],
apply_function="path.to.process_message",
)
AwaitMessageSensor is deferrable by default — same mechanism covered in the Architecture module, so a worker slot isn't held open the entire time it's waiting for a message to arrive.
The Connection
airflow connections add kafka_default \
--conn-type kafka \
--conn-extra '{"bootstrap.servers": "kafka-broker:9092", "group.id": "airflow-consumers"}'
Everything about how to reach the broker lives in this Connection, same pattern as every other external system covered in this course.
In practice, Kafka rarely appears inside the middle of a batch DAG. The far more common pattern: a standalone streaming consumer (running continuously, outside Airflow entirely - a Flink job, a small dedicated service) writes completed batches to S3/a warehouse, and that is what an Airflow DAG picks up next, often via a Dataset (covered in the previous module) rather than talking to Kafka directly at all.