# How to Set Up Kafka Connect for Streaming ETL to PostgreSQL

> Set up Kafka Connect for streaming ETL to PostgreSQL easily. Persist Kafka JSON events to relational tables automatically using the JDBC Sink Connector. Learn how now.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: how-to-guide
- Published: 2026-05-31

---

**Kafka Connect provides a scalable, fault-tolerant framework for streaming data from Kafka topics to PostgreSQL without custom consumer code, using the JDBC Sink Connector to automatically persist JSON events into relational tables.**

The DataTalksClub/data-engineering-zoomcamp repository demonstrates real-time streaming patterns using Apache Flink, but the same Kafka infrastructure supports Kafka Connect for zero-code ETL pipelines. By configuring the **JDBC Sink Connector**, you can stream JSON events from the `rides` topic directly into a PostgreSQL table for downstream analytics tools like dbt or Superset.

## Architecture Overview

The streaming ETL pipeline consists of four main components orchestrated via Docker Compose:

- **Kafka broker** (`broker`) receives JSON events on the `rides` topic via `PLAINTEXT://broker:29092` for internal communication and `PLAINTEXT_HOST://localhost:9092` for external clients.
- **Kafka Connect** (`connect`) runs as a separate container hosting the JDBC Sink Connector that transforms and loads data.
- **PostgreSQL** (`postgres`) exposes port 5432 with default credentials `postgres/postgres` and persists data to the target table.
- **Producer clients** publish ride events that Kafka Connect consumes using the `connect` group ID and writes to PostgreSQL with idempotent upserts.

## Configuring the Kafka Connect Service

Add the Connect service to your [`docker-compose.yml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/docker-compose.yml) file located at [`07-streaming/extras/python/docker/kafka/docker-compose.yml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/07-streaming/extras/python/docker/kafka/docker-compose.yml). The service must declare dependencies on both the broker and PostgreSQL to ensure proper startup order.

```yaml
services:
  connect:
    image: confluentinc/cp-kafka-connect:7.2.0
    hostname: connect
    container_name: connect
    depends_on:
      - broker
      - postgres
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: broker:29092
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_GROUP_ID: "connect-cluster"
      CONNECT_CONFIG_STORAGE_TOPIC: "_connect-configs"
      CONNECT_OFFSET_STORAGE_TOPIC: "_connect-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "_connect-status"
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_LOG4J_ROOT_LOGLEVEL: "INFO"
    volumes:
      - ./postgres-sink.json:/etc/kafka-connect/postgres-sink.json

```

The `CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"` setting is critical when working with plain JSON payloads that do not include Schema Registry metadata. The volume mount makes your connector configuration available inside the container at [`/etc/kafka-connect/postgres-sink.json`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main//etc/kafka-connect/postgres-sink.json).

## Deploying the JDBC Sink Connector

Create a [`postgres-sink.json`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/postgres-sink.json) file in the same directory as your Docker Compose configuration. This JSON payload defines how Kafka Connect maps the `rides` topic to the PostgreSQL table.

```json
{
  "name": "postgres-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "rides",
    "connection.url": "jdbc:postgresql://postgres:5432/postgres",
    "connection.user": "postgres",
    "connection.password": "postgres",
    "auto.create": "true",
    "insert.mode": "upsert",
    "pk.mode": "record_value",
    "pk.fields": "PULocationID,DOLocationID,pickup_datetime",
    "table.name.format": "processed_events",
    "batch.size": "1000",
    "dialect.name": "PostgreSqlDialect"
  }
}

```

Key configuration parameters include:

- **auto.create**: Automatically creates the target table if it does not exist, matching the schema inferred from JSON payloads.
- **insert.mode**: Set to `upsert` to handle duplicate keys gracefully using PostgreSQL's `INSERT ... ON CONFLICT` syntax.
- **pk.mode**: Uses `record_value` to extract primary key fields from the JSON payload rather than Kafka metadata.
- **pk.fields**: Defines the composite primary key using `PULocationID`, `DOLocationID`, and `pickup_datetime` to ensure uniqueness.

## Starting the Pipeline

Initialize the complete stack including the new Connect service:

```bash
docker compose up -d

```

Once all services report healthy status, deploy the connector via the Kafka Connect REST API exposed on port 8083:

```bash
curl -X POST -H "Content-Type: application/json" \
     --data @postgres-sink.json \
     http://localhost:8083/connectors

```

Verify the connector is running and check for task errors:

```bash
curl http://localhost:8083/connectors/postgres-sink/status

```

## Producing Test Events

Use a Python producer to generate test data matching the schema expected by the sink. The producer serializes dictionaries to JSON and sends them to the `rides` topic via `localhost:9092`.

```python
from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

while True:
    ride = {
        "PULocationID": 1,
        "DOLocationID": 2,
        "trip_distance": 3.5,
        "total_amount": 12.75,
        "pickup_datetime": "2024-01-01T12:00:00Z"
    }
    producer.send("rides", value=ride)
    producer.flush()
    time.sleep(1)

```

Ensure the field names in your JSON payload match the column names expected by the sink connector. The [`pass_through_job.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/pass_through_job.py) file in the workshop directory demonstrates the target schema used by the Flink sink, which aligns with the Connect configuration above.

## Verifying Data Flow

Connect to the PostgreSQL container and query the target table to confirm the streaming ETL is functioning:

```sql
SELECT * FROM processed_events LIMIT 5;

```

You should see JSON events from the `rides` topic materialized as relational rows. The connector continuously consumes from Kafka, maintaining its position in the `_connect-offsets` internal topic for fault tolerance across restarts.

## Summary

- **Kafka Connect** eliminates the need for custom consumer code when streaming JSON data from Kafka to PostgreSQL using the JDBC Sink Connector.
- The `confluentinc/cp-kafka-connect:7.2.0` image provides the runtime environment, configured via environment variables in [`docker-compose.yml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/docker-compose.yml) at [`07-streaming/extras/python/docker/kafka/docker-compose.yml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/07-streaming/extras/python/docker/kafka/docker-compose.yml).
- The **JDBC Sink Connector** supports `auto.create` for schema management and `upsert` mode for idempotent writes based on composite primary keys defined in `pk.fields`.
- Deployment requires posting a JSON configuration to the Connect REST API at `localhost:8083`, with the connector reading from the `rides` topic and writing to the `processed_events` table.

## Frequently Asked Questions

### How does Kafka Connect handle schema evolution with PostgreSQL?

The JDBC Sink Connector supports schema evolution through the `auto.evolve` setting, which allows the connector to add new columns to PostgreSQL tables when the JSON payload structure changes. For production environments, disable `auto.create` and manage schema changes manually using DDL statements to ensure data type compatibility and indexing strategies remain optimal.

### Can I achieve exactly-once delivery guarantees with this setup?

Yes, the JDBC Sink Connector supports exactly-once semantics when configured with `insert.mode: upsert` and proper primary key definitions in `pk.fields`. The connector uses idempotent writes via PostgreSQL's `ON CONFLICT` clause, ensuring that reprocessing the same Kafka offset due to failures does not create duplicate records in the database.

### What is the performance impact of the batch.size configuration?

The `batch.size` parameter (default 1000) controls how many records Kafka Connect buffers before executing a bulk insert to PostgreSQL. Increasing this value improves throughput by reducing network round-trips, but increases latency and memory usage. For high-throughput streaming ETL, monitor the consumer lag metrics and adjust the batch size based on your latency requirements and PostgreSQL's ability to handle large transactions.

### How do I troubleshoot connector failures in the Zoomcamp environment?

Check the Connect container logs using `docker logs connect` to identify connectivity issues or SQL exceptions. Verify that the `connection.url` uses the Docker service name `postgres` rather than `localhost`, and ensure the `rides` topic exists before deploying the connector. The `client.properties` file in the cohorts directory provides reference configurations for bootstrap server settings when debugging external connectivity.