Designing Kafka Topics for Streaming Taxi Data: A Production Guide

When designing Kafka topics for streaming taxi data, separate ride types into distinct topics, use stable keys like vendor_id for partitioning, enforce schemas to prevent drift, and configure SASL/SSL security with checkpointed consumer groups to ensure scalable, ordered, and fault-tolerant streaming analytics.

Designing Kafka topics for streaming taxi data requires careful consideration of granularity, partitioning strategy, and schema evolution to handle high-volume event streams effectively. The DataTalksClub/data-engineering-zoomcamp repository demonstrates production-ready patterns for NYC taxi ride events, showing how to structure topics that isolate latency, maintain per-vendor ordering, and support downstream Spark Structured Streaming aggregations.

Topic Granularity: Isolate Streams by Ride Type

Creating separate topics for each ride type keeps schemas simple and allows independent consumer groups to process only relevant data. The Zoom-Camp implementation defines distinct topics for different taxi categories in cohorts/2023/week_6_stream_processing/settings.py【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/settings.py#L7-L9】:

GREEN_TAXI_TOPIC = 'green_taxi_rides'
FHV_TAXI_TOPIC   = 'fhv_taxi_rides'
RIDES_TOPIC      = 'all_rides'

This separation prevents schema conflicts between green taxis and for-hire vehicles (FHV) while enabling a unified RIDES_TOPIC for downstream aggregation. Use lower-case snake_case with clear domain prefixes to improve discoverability and avoid naming collisions across environments.

Key Selection and Partitioning Strategy

Choosing a stable partition key guarantees ordering per key and distributes load evenly across partitions. The producer implementation in producer_confluent.py extracts the key from the first column (row[0])—typically vendor_id or dispatching_base_num—and passes it explicitly to the producer【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/producer_confluent.py#L20-L26】:

key = row[0]  # vendor_id or dispatching_base_num

producer.produce(
    topic=kafka_topic,
    key=key,           # Ensures ordering per vendor

    value=record_json,
    callback=delivery_report
)

Key-based partitioning ensures all events for the same taxi vendor land in the same partition, maintaining temporal order for time-windowed analytics while enabling horizontal scaling across multiple partitions.

Schema Enforcement and Evolution

Define a shared schema upfront to prevent drift and simplify downstream joins. The ALL_RIDE_SCHEMA in settings.py standardizes pickup and dropoff location IDs【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/settings.py#L11-L14】:

ALL_RIDE_SCHEMA = T.StructType([
    T.StructField('PUlocationID', T.StringType()),
    T.StructField('DOlocationID', T.StringType())
])

When reading the aggregated RIDES_TOPIC in streaming_confluent.py, the Spark job applies this schema immediately after ingestion【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/streaming_confluent.py#L90-L93】, ensuring that location-based aggregations never fail due to missing or mistyped fields.

Security and Authentication

Protect sensitive ride data using SASL/SSL with per-topic ACLs. Both the Spark reader and Confluent producer configure security protocols using credentials centralized in settings.py. The Spark Structured Streaming job in streaming_confluent.py sets the security context【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/streaming_confluent.py#L17-L21】:

.option('kafka.security.protocol', 'SASL_SSL')
.option('kafka.sasl.mechanism', 'PLAIN')
.option('kafka.sasl.jaas.config',
        f'''org.apache.kafka.common.security.plain.PlainLoginModule required 
        username="{CONFLUENT_CLOUD_CONFIG['sasl.username']}" 
        password="{CONFLUENT_CLOUD_CONFIG['sasl.password']}";''')

Enable idempotent writes by setting enable.idempotence=true (default in modern confluent_kafka producers when supported by the broker) to prevent duplicate records during network retries.

Consumer Group Design and Checkpointing

Each logical processing stage should use its own consumer group ID to guarantee exactly-once semantics per stage and isolate failures. The Zoom-Camp Spark jobs implement this through independent checkpoint directories:

df.writeStream
  .format('kafka')
  .option('topic', RIDES_TOPIC)
  .option('checkpointLocation', 'checkpoint')  # Unique per query

  .outputMode('append')
  .start()

The console sink in streaming_confluent.py uses a 5-second trigger (trigger(processingTime='5 seconds')) to handle back-pressure and keep the pipeline stable under bursty loads【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/streaming_confluent.py#L46-L50】.

Partition Count and Retention Policies

Match partition count to expected throughput and consumer parallelism. While the repository uses broker defaults (typically 3 partitions), production deployments should calculate based on ingest rate and consumer lag. Configure retention.ms to bound storage growth for long-running analytics, and enable log compaction on lookup tables or dimension data that require latest-state semantics per key.

Summary

• Separate topics by ride type (green, FHV) to isolate schemas and consumer groups, using settings.py to centralize topic names.
• Partition by stable keys like vendor_id to maintain ordering and distribute load, implemented in producer_confluent.py by extracting keys from CSV records.
• Enforce schemas via ALL_RIDE_SCHEMA to prevent breaking changes in downstream Spark aggregations.
• Secure the pipeline with SASL/SSL authentication and idempotent producers configured in streaming_confluent.py.
• Use independent checkpoint locations for each Spark Structured Streaming query to ensure fault tolerance and exactly-once processing semantics.

Frequently Asked Questions

How many partitions should I create for taxi data topics?

Start with a partition count equal to your expected peak consumer parallelism—typically 3-6 for development and 12-24 for production high-throughput scenarios. While the DataTalksClub example relies on broker defaults, you should scale partitions based on your ingest rate (messages per second) and the number of concurrent consumers in your group. Remember that increasing partitions afterwards requires careful rebalancing and can invalidate keyed ordering guarantees temporarily.

Should I use a single topic for all ride types or separate topics?

Use separate topics for distinct ride types (green taxi, FHV, yellow taxi) as demonstrated in the Zoom-Camp repository with GREEN_TAXI_TOPIC and FHV_TAXI_TOPIC. This隔离 isolates schema changes, allows ride-type-specific consumer groups, and prevents slow consumers of one data type from blocking others. Create a unified RIDES_TOPIC only for aggregated analytics that require a combined view, using a consistent schema like ALL_RIDE_SCHEMA to ensure compatibility.

How do I prevent duplicate taxi ride records during producer retries?

Enable idempotent writes by ensuring your producer configuration sets enable.idempotence=true. The Confluent Python client defaults to this mode when the broker supports it (Kafka 0.11+), assigning a unique producer ID and sequence numbers to each batch. This guarantees that retry attempts due to network timeouts or broker elections do not result in duplicate messages in your taxi ride topics, maintaining accurate counts for revenue and trip analytics.

What is the best way to handle schema changes in streaming taxi data?

Define and enforce a strict schema server-side using a schema registry, or client-side as shown in settings.py with ALL_RIDE_SCHEMA. When reading from Kafka in Spark, apply the schema immediately using from_json() to fail fast on incompatible changes rather than corrupting your aggregations. For evolution, use schema-compatible changes only—adding optional fields is safe, but removing or retyping columns requires versioning your topic or creating a new stream with a migration path.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}