Postgres to Databricks: Hands-On Change Data Capture Into Unity Catalog

The rows are created in Postgres. The notebooks, dashboards and models that want them run in Databricks. Whatever sits between the two decides how stale the data is, whether an update or a delete ever arrives, and how many systems somebody has to keep alive.

The usual ways to bridge that gap:

• Export and load. COPY each table to a cloud bucket, then pick the files up with COPY INTO or Auto Loader. Freshness is hours to daily, and updates and deletes mean full reloads or a merge you write yourself.
• Debezium and Kafka, read with Structured Streaming. Changes arrive within seconds. In exchange you operate Kafka, Kafka Connect, Debezium and a running Databricks job, and the merge logic is yours.
• Lakeflow Connect. The PostgreSQL connector from Databricks does change data capture (CDC) on Databricks serverless compute, with nothing to host. The rows land in streaming tables governed by Unity Catalog.

This tutorial takes a fourth route. OLake Go, an open source (Apache 2.0) replication tool, reads the Postgres write-ahead log (WAL) through logical replication and commits the changes as Apache Iceberg tables into Unity Catalog. Databricks treats them as managed Iceberg tables, and DuckDB and Spark can read the very same files.

I ran every step against a Databricks workspace on Google Cloud, with Postgres 16 and OLake Go v1.1.0. The first sync finished in 37 seconds and the change sync in 28 seconds. The figures in this post are the ones that run produced.

What you need

• Postgres 10 or later. Mine was a database called shop with two tables, customers and orders.
• A Databricks workspace with Unity Catalog, and an admin who can turn on external data access.
• Docker, for OLake UI, and about 30 minutes.

Step 1: Prepare Postgres for CDC

Start with the server settings. wal_level only takes effect after a restart:

ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;
ALTER SYSTEM SET wal_sender_timeout = 0;   -- so a long first snapshot is not cut off

On Amazon Relational Database Service (RDS), Aurora, Azure or Cloud SQL, wal_level is a parameter group or server setting instead.

To follow along with my data, create the two tables and fill them. Skip this block if you already have tables:

CREATE TABLE customers (
  id SERIAL PRIMARY KEY, name TEXT, email TEXT, city TEXT, tier TEXT,
  created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now());

CREATE TABLE orders (
  id SERIAL PRIMARY KEY, customer_id INT REFERENCES customers(id), product TEXT,
  amount NUMERIC(10,2), status TEXT,
  created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now());

-- 8 customers and 40 orders
INSERT INTO customers (name, email, city, tier) VALUES
  ('Aarav Mehta','[e-mail]','Mumbai','gold'),   ('Priya Sharma','[e-mail]','Bengaluru','silver'),
  ('Liam Carter','[e-mail]','Austin','bronze'),  ('Sofia Rossi','[e-mail]','Milan','gold'),
  ('Kenji Tanaka','[e-mail]','Tokyo','silver'), ('Emma Schulz','[e-mail]','Berlin','bronze'),
  ('Noah Kim','[e-mail]','Seoul','gold'),        ('Zara Ali','[e-mail]','Dubai','silver');

INSERT INTO orders (customer_id, product, amount, status)
SELECT 1 + (n % 8), (ARRAY['Laptop','Headphones','Monitor','Keyboard','Mouse','Webcam'])[1 + n % 6],
       ROUND(20 + ((n * 7919) % 120000) / 100.0, 2), (ARRAY['placed','shipped','delivered'])[1 + n % 3]
FROM generate_series(1, 40) AS n;

Now the role OLake Go logs in as, a publication for the two tables and a pgoutput replication slot:

CREATE ROLE olake WITH LOGIN REPLICATION PASSWORD '<password>';
GRANT USAGE ON SCHEMA public TO olake;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO olake;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO olake;

CREATE PUBLICATION olake_pub FOR TABLE customers, orders;
SELECT pg_create_logical_replication_slot('olake_slot', 'pgoutput');

-- check the result
SHOW wal_level;                                        -- logical
SELECT slot_name, plugin, active FROM pg_replication_slots;
SELECT pubname, puballtables FROM pg_publication;

Things to carry over to a production database:

• Primary keys. A table without one needs ALTER TABLE <table> REPLICA IDENTITY FULL; so that its updates and deletes carry the old row.
• A slot and a publication of its own for each job. Two jobs on a shared slot read each other's changes, and both lose data.
• WAL on disk. The slot holds WAL until OLake Go reads it. Set max_slot_wal_keep_size so a stalled slot is invalidated before the disk fills, and alert on pg_replication_slots.wal_status.
• RDS and Aurora. The REPLICATION attribute is reserved there, so grant the rds_replication role instead.

Step 2: Set up Unity Catalog

OLake Go writes through Unity Catalog's Iceberg representational state transfer (REST) endpoint, and Databricks allows writes through that endpoint for managed Iceberg tables only (Databricks: access tables from Iceberg clients). Four things have to be in place in the workspace, once:

1. External data access on the metastore. In Catalog, click the gear icon, open Metastore and turn on External data access. While it is off, the endpoint turns away any client other than Databricks itself.
2. A catalog with managed storage. OLake Go creates one schema per job inside it.
3. EXTERNAL USE SCHEMA for the principal that writes. Unity Catalog checks it before it serves a table to an outside engine.
4. A personal access token. Create it under Settings, Developer, Access tokens.

Run this in the Databricks SQL editor, with your own principal:

-- The catalog OLake Go writes into. Its managed storage holds the Iceberg files.
CREATE CATALOG IF NOT EXISTS lakehouse MANAGED LOCATION '<external location URL>/lakehouse';

-- What the writing principal needs: browse and create schemas, then read and write tables in them
GRANT USE CATALOG, CREATE SCHEMA ON CATALOG lakehouse TO `[e-mail]`;
GRANT USE SCHEMA, CREATE TABLE, SELECT, MODIFY ON CATALOG lakehouse TO `[e-mail]`;
GRANT EXTERNAL USE SCHEMA ON CATALOG lakehouse TO `[e-mail]`;

My catalog was named datazip and my token expired after seven days. The snippets here use lakehouse, so swap in your own catalog name.

⚠️ EXTERNAL USE SCHEMA is not part of ALL PRIVILEGES, and only the catalog owner can grant it (Databricks: Unity Catalog privileges). Grant it explicitly, even when the principal owns the catalog.

Step 3: Start OLake UI and connect Postgres

curl -sSL https://raw.githubusercontent.com/datazip-inc/olake-ui/master/docker-compose-v1.yml | docker compose -f - up -d

Open http://localhost:8000 and sign in with admin / password.

Open Sources, click Create Source and pick Postgres. Under Change data capture (CDC), leave Enable CDC on, then enter olake_slot as the Replication Slot and olake_pub as the Publication. The default Initial Wait Time is fine.

Under Connection details, fill in the host, port, database name, username and password. My Postgres ran in another container, so the host was host.docker.internal, with database shop and user olake.

Click Set up source. OLake Go tests the connection and then checks wal_level, the replication privilege, the slot and the publication before anything is saved. If one fails, the dialog shows the current value next to the recommended one.

Step 4: Add the Unity Catalog destination

Open Destinations and click Create Destination. Pick Apache Iceberg as the connector, Unity as the catalog type and Personal Access Token as the authentication type. Then fill in:

Field	Value
REST Catalog URL	https://<workspace-url>/api/2.1/unity-catalog/iceberg-rest
Unity Catalog Name	lakehouse, the catalog from step 2
Token	the token from step 2
Catalog Name	olake_iceberg, the default when left empty

Click Create. OLake Go calls the Unity Catalog endpoint with the token and saves the destination.

A permission error here usually points back to step 2: External data access is off, the principal lacks EXTERNAL USE SCHEMA, or the catalog name is wrong.

Step 5: Create the job and run the first sync

1. Open Jobs and click Create Job. Name it postgres_to_databricks and give it a schedule.
2. Select the source and the destination you just created.
3. Under Advanced Settings, set Target Query Engines to Databricks, plus DuckDB and Apache Spark if they will read the tables too.
4. Click Next. On the Streams page, tick customers and orders, set the sync mode to Full Refresh + CDC and the ingestion mode to Upsert.
5. Leave the Upsert Type as OLake Go resolved it, then click Create Job and choose Sync Now from the job's menu.

The target engines setting matters here. Under Unity Catalog, Databricks reads Iceberg v3 deletion vectors and rejects commits that carry equality deletes. With Databricks in the list, OLake Go set Deletion Vector on both streams by itself, so I had no format to choose.

In my run the first sync copied 8 customers and 40 orders (48 records, 2.23 KB) in 37 seconds, container start included. It records the current WAL position and then reads each table in parallel chunks, so a row changed during the copy is not lost.

OLake Go names the Unity Catalog schema after the job, the source database and the source schema, which gave me postgres_to_databricks_shop_public. The Iceberg DB field at the top of the Streams page changes that name before the first sync.

Step 6: Query the tables in Databricks

There is nothing to register and no external table to declare. The SQL editor or a notebook reads the tables by their three-part name:

-- What OLake Go created (schema = <job name>_<source database>[_<source schema>])
SHOW TABLES IN lakehouse.postgres_to_databricks_shop_public;

SELECT id, name, city, tier
FROM lakehouse.postgres_to_databricks_shop_public.customers
ORDER BY id;

SELECT status, count(*) AS n, round(sum(amount), 2) AS total
FROM lakehouse.postgres_to_databricks_shop_public.orders
GROUP BY status ORDER BY status;

I used a 2X-Small serverless SQL warehouse, which is enough for checks like these.

Each table carries a few extra columns from OLake Go: _olake_id, _op_type, _olake_timestamp and _cdc_timestamp. Postgres tables add _cdc_lsn, the WAL position each changed row came from.

Step 7: Change rows in Postgres and sync again

A snapshot is the easy half. Here are the 12 changes I made between the two syncs:

INSERT INTO customers (name, email, city, tier) VALUES
  ('Maya Patel', '[e-mail]', 'Pune', 'silver'),
  ('Lucas Silva', '[e-mail]', 'São Paulo', 'bronze');
INSERT INTO orders (customer_id, product, amount, status) VALUES
  (9, 'Laptop', 1299.00, 'placed'), (9, 'Mouse', 25.50, 'placed'),
  (10, 'Monitor', 349.99, 'placed'), (3, 'Keyboard', 89.00, 'placed'), (7, 'Webcam', 79.90, 'placed');
UPDATE customers SET tier = 'gold',      updated_at = NOW() WHERE name = 'Liam Carter';
UPDATE customers SET city = 'Abu Dhabi', updated_at = NOW() WHERE name = 'Zara Ali';
UPDATE orders    SET status = 'delivered', updated_at = NOW() WHERE id IN (1, 4, 7);

Choose Sync Now again. The second sync read 12 records (594 bytes) from the slot and finished in 28 seconds.

For customers it wrote one new data file with 4 rows: the 2 inserts plus the new versions of the 2 updated rows. One deletion vector, a small bitmap that marks replaced positions in a data file, retires the 2 old versions. The commit report in the sync log reads addedDataFiles=1, addedRecords=4, addedDVs=1 and addedPositionalDeletes=2.

The orders table has the same shape with 8 rows and 3 deleted positions, and the equality delete count stays at 0. Run the SELECT from step 6 again, with no refresh first. Liam Carter is gold, Zara Ali has moved to Abu Dhabi, and the two new customers are listed.

The _op_type column says how each row arrived: r for a snapshot row, c for an insert, u for an update and d for a delete. The orders roll-up matched Postgres to the cent, with 16 delivered orders for 8659.68, 18 placed for 10522.26 and 11 shipped for 6397.25.

Schema changes. Add a column in Postgres and OLake Go adds it to the Iceberg table on the next sync, then commits the new schema to Unity Catalog. Widening changes such as int to bigint or float to double are applied, while narrowing or incompatible type changes fail the sync rather than corrupt the table.

Read the same tables from DuckDB

No export happens here. DuckDB attaches the catalog as an Iceberg REST catalog with the same token, and Unity Catalog hands it short-lived storage credentials per table.

INSTALL iceberg; LOAD iceberg;
INSTALL httpfs;  LOAD httpfs;
CREATE SECRET unity (TYPE iceberg, TOKEN '<personal access token>');
ATTACH 'lakehouse' AS uc (
  TYPE iceberg,
  ENDPOINT 'https://<workspace-url>/api/2.1/unity-catalog/iceberg-rest',
  SECRET unity,
  ACCESS_DELEGATION_MODE 'vended_credentials'
);
SELECT id, name, city, tier
FROM uc.postgres_to_databricks_shop_public.customers
ORDER BY id;

DuckDB 1.5.5 returned the same 10 customers and the same orders roll-up as Databricks. Spark 4.0.1 with Iceberg 1.10.0 did as well. The principal behind the token needs EXTERNAL USE SCHEMA for this too, the same as OLake Go.

Data types to know about

Postgres	Databricks shows	Note
int, serial, smallint	INT
bigint, bigserial	BIGINT
numeric, decimal, real	DOUBLE	round at query time when you need exact decimal arithmetic
date, timestamp, timestamptz	TIMESTAMP	Coordinated Universal Time (UTC); a date gains a midnight time
json, jsonb, uuid, arrays	STRING	parse jsonb at query time

Tips and limits

• Check predictive optimization. Unity Catalog runs snapshot expiration and file compaction on managed Iceberg tables when it is on (Databricks: Iceberg tables).
• Rotate the token. An expired personal access token fails the next sync at the first catalog call. For production jobs, use a service principal with OAuth machine-to-machine credentials.
• A null-path error means the wrong delete format. Can not create a Path from a null string [ErrorCode: 2015] is what Unity Catalog returns when a client commits equality delete files. Set Databricks as a target query engine, and start that table again from a full refresh.
• No primary key, no updates. If a table has no primary key and cannot get REPLICA IDENTITY FULL, inserts replicate but updates and deletes do not.
• These are Iceberg tables, not Delta tables. If a pipeline needs native Delta Lake features on these rows, load them into a Delta table downstream.

How the methods compare

Method	Freshness	Updates and deletes	Where the data lands	What you run
Lakeflow Connect PostgreSQL connector	CDC, run by Databricks	yes, through CDC	streaming tables governed by Unity Catalog	nothing to host; Databricks serverless compute
Debezium, Kafka and Structured Streaming	seconds	yes, with your own merge logic	Kafka topics, then tables written by a Databricks streaming job	Kafka, Kafka Connect, Debezium and a running Databricks job
COPY to a bucket, then COPY INTO or Auto Loader	hours to daily	no; full reloads or your own merge	tables written by a Databricks job	an export script, a bucket and a Databricks job
OLake Go to Unity Catalog (this post)	CDC on a schedule down to one minute	yes, as deletion vectors	Iceberg tables that Databricks, DuckDB and Spark all read	one Docker Compose stack, open source; your storage and your own compute

When Databricks is the only reader and you would rather have Databricks run the pipeline, Lakeflow Connect is the simpler choice. Writing Iceberg once starts to pay off when another engine needs the same rows, because there is no second copy to make.

Where to go from here

• Point the source at your real database. On RDS, Aurora, Azure and Cloud SQL the main difference is where wal_level is set.
• The full guide has the Spark connection, the command line setup and the complete type mapping: Postgres to Databricks on olake.io.

If a step behaves differently on your workspace, post the exact message in the comments and I will look into it.