# Designing Kafka Topics for Streaming Taxi Data: A Production Guide

> Design Kafka topics for streaming taxi data. Learn production best practices for partitioning, schema enforcement, and security to build scalable, fault-tolerant analytics.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: best-practices
- Published: 2026-05-30

---

**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)【https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/settings.py#L7-L9】:

```python
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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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】:

```python
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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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】:

```python
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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/settings.py). The Spark Structured Streaming job in [`streaming_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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】:

```python
.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:

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

  .outputMode('append')
  .start()

```

The console sink in [`streaming_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/settings.py) to centralize topic names.
- **Partition by stable keys** like `vendor_id` to maintain ordering and distribute load, implemented in [`producer_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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.