How to stream PostgreSQL CDC into Databricks with Debezium

[Victoria_Bukta]

Data stored in operational databases powers critical business decisions, but making it available in the lakehouse presents significant architecture challenges. Strict enterprise environments could have constraints such as private corporate networks, customer-controlled environments, or restricted subnets; network policies strictly prohibit inbound connections. A traditional pull-based ingestion model in which the lakehouse directly queries an internal database is simply not feasible.

To solve this, engineers at NVIDIA and Databricks teamed up on a push-based architecture that keeps capture under the control of the database team, and contributed it to the open-source Debezium project. Databricks built the Zerobus Ingest sink for Debezium Server, and NVIDIA contributed envelope mode, which lands the full Debezium change event along with delivery safeguards such as idempotency keys.

In this model, a sidecar process running locally within the database network tail-reads the binlog, write-ahead log (WAL) or streams change events of the database and pushes the data to Zerobus Ingest, which writes data directly to governed Delta tables in Databricks. With just outbound network connectivity, you can stream change events with Zerobus, without needing to set up additional infrastructure.

Once you create the destination table and configure Debezium Server, launching it starts streaming change events into Databricks.

For the architecture behind it, including when a push model fits and how envelope mode works, read the companion overview on the Databricks blog.

In this post, we will show you how you can get this up and running end-to-end with a Postgres database, resulting in a type-1 dimensional table, representing the current state of your database table.

How-to tutorial:  a self-contained toy example

The Zerobus sink can write change events in two shapes, and you pick one per target table:

• Typed (payload.mode=typed, the default) maps each change event onto a table with the source’s own columns (id, name, balance) and writes JSON. The event shape must match that schema, and deletes are not applied for you.
• Envelope (payload.mode=envelope, contributed by NVIDIA) lands the whole Debezium change event in a fixed nine-column table: target_table, destination, partition, operation, idempotency_key, key, value, source_position, headers. The full before/after row images, operation and source position are kept, so current-state and history tables can both be built downstream. Records can be sent as JSON or Protobuf.

The pipeline we'll build:

Local Postgres  ──▶  Debezium Server (Zerobus sink)  ──▶  
Zerobus Ingest  ──▶  Delta table  ──▶  current-state table

1. Create and seed a sample PostgreSQL table

Start a disposable PostgreSQL 16 with logical replication enabled:

export PG_PASSWORD='<postgres-password>'

docker run -d --name dbz-zerobus-pg \
  -e POSTGRES_PASSWORD="$PG_PASSWORD" -e POSTGRES_DB=shop \
  -p 127.0.0.1:5432:5432 postgres:16 \
  -c wal_level=logical -c max_replication_slots=4 -c max_wal_senders=4

Create the table, publication and one baseline row:

CREATE TABLE public.customers (
  id      INT PRIMARY KEY,
  name    TEXT,
  balance INT
);
ALTER TABLE public.customers REPLICA IDENTITY FULL;
CREATE PUBLICATION customers_pub FOR TABLE public.customers;
INSERT INTO public.customers VALUES (1, 'Snapshot baseline', 5);

REPLICA IDENTITY FULL makes update and delete events carry the full previous row. The publication limits capture to this table, and Debezium creates its replication slot on first start. The integer balance keeps the JSON readable, because PostgreSQL NUMERIC uses a connector-specific decimal encoding. The baseline row arrives as a snapshot read (op = r). The change cycle in 7.6 uses a separate key, 123.

2. Configure Debezium Server with the Zerobus sink

Debezium Server is configured through a single application.properties. The source half (PostgreSQL) is standard Debezium; the sink half targets Zerobus.

# ---- Source: PostgreSQL ----
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.plugin.name=pgoutput
debezium.source.database.hostname=127.0.0.1
debezium.source.database.port=5432
debezium.source.database.user=postgres
debezium.source.database.password=${PG_PASSWORD}
debezium.source.database.dbname=shop
debezium.source.topic.prefix=shop
debezium.source.table.include.list=public.customers
debezium.source.publication.name=customers_pub
debezium.source.publication.autocreate.mode=disabled
debezium.source.slot.name=customers_zerobus_json
debezium.source.slot.drop.on.stop=true
debezium.source.snapshot.mode=initial
debezium.source.offset.storage.file.filename=data/offsets-json.dat
debezium.source.offset.flush.interval.ms=1000

# ---- Event format handed to the sink ----
debezium.format.key=json
debezium.format.value=json
debezium.format.key.schemas.enable=false
debezium.format.value.schemas.enable=false

# ---- Sink: Databricks Zerobus (gRPC) ----
debezium.sink.type=zerobus
debezium.sink.zerobus.endpoint=<workspace-id>.zerobus.<region>.cloud.databricks.com
debezium.sink.zerobus.workspace.url=https://<workspace-host>
debezium.sink.zerobus.client.id=<service-principal-application-id>
debezium.sink.zerobus.client.secret=${ZEROBUS_CLIENT_SECRET}
debezium.sink.zerobus.table=main.cdc.customers_events
debezium.sink.zerobus.payload.mode=envelope
debezium.sink.zerobus.record.format=json
debezium.sink.zerobus.json.flexible.fields.encoding=string
debezium.sink.zerobus.idempotency.mode=source
debezium.sink.zerobus.tombstone.handling.mode=event
debezium.sink.zerobus.max.record.bytes=10000000
quarkus.kubernetes-config.enabled=false

Secrets are ${…} expressions read from the environment, so they never appear in the file. The less obvious settings:

• No unwrap transform (such as ExtractNewRecordState): value must keep the full Debezium event.
• json.flexible.fields.encoding=string sends key, value, source_position and headers as JSON text, which Zerobus stores in VARIANT columns. Sending nested objects instead fails with Zerobus error 4044.
• idempotency.mode=source writes the deterministic idempotency_key.
• tombstone.handling.mode=event keeps Debezium’s post-delete tombstone as its own row; drop skips it.
• max.record.bytes rejects an oversized record before ingestion, so its offset is never committed.
• slot.drop.on.stop=true and the postgres user are for this demo only. In production, keep the slot so capture can resume, and use a dedicated role with REPLICATION and SELECT.
• Authenticate with an OAuth service principal; a personal access token is not a Zerobus credential.

3. Choose a delivery route

This walkthrough uses the gRPC route, debezium.sink.type=zerobus. It holds a persistent stream to Zerobus Ingest and carries envelope mode in both JSON and Protobuf. It’s the route we tested end to end.

Zerobus Ingest offers other ingestion interfaces as well; see the Zerobus Ingest documentation. If you use another route, validate its authentication and payload contract separately.

4. Create the destination table in Databricks

CREATE TABLE main.cdc.customers_events (
  target_table     STRING,
  destination      STRING,
  partition        INT,
  operation        STRING,
  idempotency_key  STRING,
  key              VARIANT,
  value            VARIANT,
  source_position  VARIANT,
  headers          VARIANT
) USING DELTA;

In envelope mode, the target table has these nine columns whatever the source table looks like:

Column	Holds
target_table	The Unity Catalog table the record was routed to
destination	The Debezium destination, e.g. shop.public.customers
partition	The Debezium partition number
operation	read, create, update, delete or tombstone
idempotency_key	sha256:… identity of the event, stable across redelivery
key	The record key, e.g. {"id": 123}
value	The full Debezium event (before, after, op, source, ts_ms); null for a tombstone
source_position	Source partition and offset, including the PostgreSQL LSN and transaction ID
headers	Debezium record headers

Query the VARIANT columns with variant_get. Zerobus doesn’t create or alter tables, so create this one in advance. A new source column simply shows up inside value; the raw table never needs an ALTER. For Protobuf, create a separate table with the same columns but STRING for key, value, source_position and headers.

Grant the service principal that Debezium authenticates as access to the table:

GRANT USE CATALOG ON CATALOG main TO `<service-principal-application-id>`;
GRANT USE SCHEMA  ON SCHEMA  main.cdc TO `<service-principal-application-id>`;
GRANT SELECT, MODIFY ON TABLE main.cdc.customers_events TO `<service-principal-application-id>`;

The service principal also needs workspace access and an OAuth secret. The Zerobus Ingest setup guide covers both, and how to find your endpoint.

5. Launch the connector

Now that the source and destination tables are set up, we are ready to launch the connector!

The Zerobus sink and its envelope mode ship in Debezium Server 3.7.0.Final. The standard distribution already includes the sink, the Zerobus Ingest SDK and the PostgreSQL connector, so there’s nothing to build. Download and extract it:

curl -LO https://repo1.maven.org/maven2/io/debezium/debezium-server-dist/3.7.0.Final/debezium-server-dist-3.7.0.Final.tar.gz
tar -xzf debezium-server-dist-3.7.0.Final.tar.gz
cd debezium-server

Save the configuration from 7.2 as config/application.properties. Then start Debezium Server with Java 21 or later, with the two secrets supplied as environment variables:

export ZEROBUS_CLIENT_SECRET='<service-principal-secret>'
mkdir -p data
./run.sh

On startup, Debezium creates the replication slot, snapshots public.customers, switches to streaming and opens a Zerobus stream for the target table. To confirm the pipeline works end to end, query the table in Databricks. The baseline row should appear as one operation = 'read' row for id = 1:

SELECT operation, variant_get(key, '$.id', 'int') AS id
FROM main.cdc.customers_events;

6. Validate: insert → update → delete round-trip

Once the snapshot row has landed, run three separately committed statements in PostgreSQL:

INSERT INTO public.customers VALUES (123, 'Alice', 10);
UPDATE public.customers SET balance = 25 WHERE id = 123;
DELETE FROM public.customers WHERE id = 123;

Then query the change events in Databricks SQL:

SELECT
  variant_get(key,   '$.id',             'int')    AS id,
  operation,
  variant_get(value, '$.op',             'string') AS dbz_op,
  variant_get(value, '$.before.balance', 'int')    AS before_balance,
  variant_get(value, '$.after.balance',  'int')    AS after_balance,
  variant_get(value, '$.source.lsn',     'bigint') AS source_lsn,
  idempotency_key
FROM main.cdc.customers_events
WHERE variant_get(key, '$.id', 'int') = 123
ORDER BY source_lsn NULLS LAST, operation;

Results from our run on Debezium Server 3.7.0.Final (your LSNs and hashes will differ):

id	operation	dbz_op	before_balance	after_balance	source_lsn	idempotency_key
123	create	c	null	10	26793112	sha256:f31b…
123	update	u	10	25	26793296	sha256:41c5…
123	delete	d	25	null	26793448	sha256:e310…
123	tombstone	null	null	null	null	sha256:68a3…

What to check:

• Row images: the insert has only after, the update shows 10 → 25, and the delete keeps the old row in before with a null after.
• Order: source LSNs increase across the three commits. That is the order downstream processing should apply.
• Tombstone: it’s a separate row with a null value. It shares the delete’s source position but has its own idempotency_key, because the operation is part of the identity.
• Counting: delivery is at least once, so a retry can add a second physical row with the same idempotency_key. Count unique changes with count(DISTINCT idempotency_key). With the snapshot row, our table held 5 rows and 5 distinct identities.

The Debezium Server log for the same run, trimmed:

INFO  [i.d.c.p.c.PostgresReplicationConnection] Creating replication slot with command CREATE_REPLICATION_SLOT "customers_zerobus_json" LOGICAL pgoutput
INFO  [i.d.e.a.AsyncEmbeddedEngine] All tasks have started successfully.
INFO  [i.d.r.RelationalSnapshotChangeEventSource] Snapshot step 7 - Snapshotting data
INFO  [i.d.r.RelationalSnapshotChangeEventSource]   Finished exporting 1 records for table 'public.customers' (1 of 1 tables)
INFO  [i.d.p.s.AbstractSnapshotChangeEventSource] Snapshot completed
INFO  [i.d.p.ChangeEventSourceCoordinator] Starting streaming
INFO  [i.d.c.p.PostgresStreamingChangeEventSource] Processing messages
INFO  [i.d.s.d.z.ZerobusChangeConsumer] Opening Zerobus json stream for table 'main.cdc.customers_events'
INFO  [i.d.c.c.BaseSourceTask] 5 records sent during previous 00:00:10.352, last recorded offset of {server=shop} partition is {lsn_proc=26793448, ..., messageType=DELETE, ...}

A log line shows the records left Debezium; the SQL result is the proof they landed and can be queried.

7. Build the current-state (type-1) table — optional Auto CDC

The envelope table is a bronze changelog. To serve the data, a Lakeflow pipeline turns it into silver and gold. First, it normalizes each event into typed columns. Next, AUTO CDC applies the changes to a current-state table (SCD Type 1). Finally, a materialized view builds a gold summary on top. Create a serverless, triggered pipeline with main.cdc as its default catalog and schema, using this SQL:

-- Silver: typed change rows (tombstones excluded)
CREATE OR REFRESH STREAMING TABLE customers_changes AS
SELECT
  variant_get(key,   '$.id',            'int')    AS id,
  operation,
  variant_get(value, '$.after.name',    'string') AS name,
  variant_get(value, '$.after.balance', 'int')    AS balance,
  variant_get(value, '$.source.lsn',    'bigint') AS source_lsn,
  idempotency_key
FROM STREAM(main.cdc.customers_events)
WHERE operation IN ('read', 'create', 'update', 'delete');

-- Silver: current state (SCD Type 1)
CREATE OR REFRESH STREAMING TABLE customers_current;

CREATE FLOW customers_current_flow AS AUTO CDC INTO customers_current
FROM STREAM(customers_changes)
KEYS (id)
APPLY AS DELETE WHEN operation = 'delete'
SEQUENCE BY source_lsn
COLUMNS * EXCEPT (operation, source_lsn, idempotency_key)
STORED AS SCD TYPE 1;

-- Gold: business summary
CREATE OR REFRESH MATERIALIZED VIEW customer_summary AS
SELECT count(*) AS active_customers, sum(balance) AS total_balance
FROM customers_current;

After one pipeline update:

• customers_current holds only key 1 (Snapshot baseline, balance 5). Key 123 is gone because its delete was applied.
• customer_summary reports 1 active customer with a total balance of 5.
• To keep history instead, use the same flow with STORED AS SCD TYPE 2. Key 123 then keeps two versions (balance 10, then 25), each bounded by source LSNs in __START_AT and __END_AT, with the last one closed at the delete.

Sequencing uses source.lsn. In our run it ordered every change correctly, including the initial snapshot row. Don’t sequence by ts_ms or idempotency_key; neither is a source order. Because delivery is at least once, production pipelines should also decide how to drop repeated events, for example, by idempotency_key, before AUTO CDC.

Conclusion

Choosing the right CDC architecture comes down to network constraints, operational needs, and infrastructure preferences. Combining Debezium Server with Zerobus Ingest delivers a lightweight, push-based CDC solution that streams database updates directly into Delta tables with no intermediate message queues or temporary staging storage required.

Below, we have a decision matrix for you to take away to help guide you to deciding which approach fits your needs and understand the trade-offs.

Requirement / Feature	Debezium + Zerobus (Local Push)	DIY (Debezium + Kafka + Spark)
Ingestion Direction	Push-based (Source agent pushes outward)	Push-based (Source agent pushes outward to Kafka)
Network Requirements	Requires only outbound access from source network	Requires outbound access to Kafka cluster / brokers
Private / Customer Networks	Ideal fit for locked-down or private networks	Good fit for private networks, but requires managing Kafka connectivity
Kafka Infrastructure	Not required (streams directly via Zerobus)	Required (Kafka cluster acting as intermediate buffer)
Connector Management	Requires running Debezium Server locally	High operational overhead (managing Kafka, Connect, and Spark ingestion jobs)
Ecosystem Compatibility	Leverages full Debezium connector ecosystem; envelope mode lands raw change events for AUTO CDC silver and gold tables	Leverages full Debezium and Kafka Connect ecosystem with custom Spark streaming to Delta (Bronze) and AUTO CDC jobs