How to Set Up Kafka Connect for Streaming ETL to PostgreSQL

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 file located at 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.

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.

Deploying the JDBC Sink Connector

Create a 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.

{
  "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:

docker compose up -d

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

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:

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.

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 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:

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 at 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.

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