OpenMetadata + Kafka + Databricks Lineage Demo

A hands-on demo of how metadata and lineage are captured across a real CDC pipeline:

Postgres → Debezium → Kafka → Databricks Delta Lake (bronze → silver → gold), with OpenMetadata stitching every hop into a searchable knowledge graph that AI agents can query via MCP.

┌─────────────┐   WAL CDC    ┌──────────────┐   topics   ┌─────────────────────┐
│  Postgres   │ ───────────► │ Kafka Connect│ ─────────► │ Databricks Unity    │
│  ecommerce  │   Debezium   │  + Schema    │  Confluent │ Catalog medallion   │
│             │              │  Registry    │  Delta Sink│ bronze/silver/gold  │
└─────────────┘              └──────────────┘      │     └─────────────────────┘
                                       ▲           │               ▲
                                       │      S3 staging           │
                                       │           ▼               │
                              OpenMetadata ◄───────┴───────────────┘
                              (lineage + MCP)

What you need

Requirement	Notes
Docker + Compose v2	~6 GB RAM free recommended
Databricks Free Edition	Workspace with Unity Catalog (on by default)
AWS account	S3 bucket for sink staging + IAM role for Databricks

1. Databricks Free Edition setup

1. Sign up for Databricks Free Edition and open your workspace.
2. Open SQL Warehouses. Free Edition includes a serverless warehouse — start it and copy:
   • Server hostname (e.g. dbc-xxxx.cloud.databricks.com) → DATABRICKS_HOST
   • HTTP path (e.g. /sql/1.0/warehouses/<id>) → DATABRICKS_HTTP_PATH
3. Go to Settings → Developer → Access tokens → Generate new token. Copy into DATABRICKS_TOKEN.
4. Unity Catalog is already enabled on Free Edition — no extra setup.

2. AWS S3 staging bucket setup

The Confluent Databricks Delta Lake Sink stages Avro/Parquet files in S3, then Databricks runs COPY INTO from that bucket. Free Edition is serverless-only (no instance profiles), so Databricks accesses the bucket through a Unity Catalog storage credential + external location.

2a. Create the bucket and connector IAM user

# Replace names/region as needed
aws s3 mb s3://om-demo-kafka-staging --region us-east-1

1. Create an IAM user (e.g. om-demo-kafka-connect) with programmatic access.
2. Attach a policy based on databricks/iam-policy-s3-staging.json (replace YOUR_STAGING_BUCKET).
3. Put the access key ID / secret into .env as AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY.
4. Set S3_STAGING_BUCKET=om-demo-kafka-staging and DATABRICKS_S3_REGION=us-east-1.

2b. Create the IAM role Databricks will assume

1. Create an IAM role (e.g. om-demo-databricks-staging) with the same S3 policy as above.
2. Start with the trust-policy template in databricks/iam-trust-policy-databricks.json.template:
   • The Databricks UC master role ARN on AWS is typically
     arn:aws:iam::414351767826:role/unity-catalog-prod-UCMasterRole-14S5ZJCTLRC6Y
     (confirm in Databricks docs for your cloud).
   • Replace YOUR_DATABRICKS_EXTERNAL_ID, YOUR_AWS_ACCOUNT_ID, and YOUR_ROLE_NAME after Databricks shows them (step 2c). Databricks uses a self-assume pattern — you will update the trust policy once with the external ID Databricks assigns.

2c. Register storage credential + external location (Free Edition path)

In Databricks Catalog → + Add → Add a credential:

1. Choose AWS IAM role, paste the role ARN, create the credential.
2. Databricks shows an External ID — paste it into the role trust policy and update the role.
3. Click Validate on the credential.
4. Catalog → + Add → Add an external location:
   • URL: s3://om-demo-kafka-staging/ (must cover the bucket root — the sink stages under topics/<topic-name>/)
   • Storage credential: the one you just created
   • Test connection, then create.
5. Grant yourself READ FILES (and WRITE FILES if you want notebooks to write there) on the external location.

If COPY INTO fails with 403 Forbidden / AnonymousAWSCredentials, the external location or storage credential trust policy is incomplete — re-validate the credential and confirm the location covers the staging bucket.

This replaces the instance-profile setup in Confluent’s docs — Free Edition cannot attach instance profiles to compute.

2d. Create bronze / silver / gold tables

In the Databricks SQL Editor, run databricks/setup_databricks.sql.
It creates the demo catalog, bronze landing tables aligned to Debezium staging Avro, and silver/gold CTAS statements.

The Confluent sink writes short table names (customers, products, orders, order_items) and targets demo.bronze via connector settings delta.lake.catalog / delta.lake.database (from DATABRICKS_CATALOG / DATABRICKS_BRONZE_SCHEMA in .env). Create those tables first with databricks/setup_databricks.sql (auto.create is off so the sink will not invent tables under default).

If COPY INTO fails with a schema mismatch, run databricks/recreate_bronze.sql (adds __deleted, drops line_total / partition), then make connectors.

Re-run the silver/gold section after data lands to refresh aggregates (Unity Catalog records that lineage automatically).

3. Configure and start the demo

cp .env.example .env
# fill DATABRICKS_* / S3_* / AWS_* values

./demo.sh          # guided walkthrough (recommended)
# — or —
make up            # start stack
make connectors    # register Debezium + Databricks sink
make ingest        # run OpenMetadata ingestion
make lineage       # add topic → bronze edges
make live-data     # optional continuous order generator

OpenMetadata UI: http://localhost:8585 — login admin / admin.

Service	Port
OpenMetadata	8585
Kafka Connect	8083
Schema Registry	8081
Kafka (external)	9092
Postgres	5432
Elasticsearch	9200

4. What the stack runs (lean)

Container	Role
postgres	Ecommerce OLTP and OpenMetadata DB (wal_level=logical)
kafka	Single-node KRaft broker
schema-registry	Avro schemas for Connect + OM
connect	Debezium Postgres source + Confluent Databricks Delta Lake Sink
elasticsearch	Search backend for OpenMetadata (512 MB heap)
openmetadata-server	Catalog / lineage / MCP (no always-on Airflow)
datagen (profile live)	Inserts/updates orders so CDC stays live

Metadata ingestion is one-shot via openmetadata/ingestion (make ingest) to keep RAM down.

5. Demoing OpenMetadata’s feature set

After make ingest + make lineage:

1. Discovery — Search for orders, customers, or daily_revenue. Facet by service (Postgres / Kafka / Unity Catalog).
2. End-to-end lineage — Open ecommerce.public.orders → Lineage tab. You should see: Postgres table → Kafka topic → demo.bronze.orders → demo.silver.orders → demo.gold.*
3. Column-level lineage — Drill into gold tables; Unity Catalog system lineage plus CTAS gives column edges on Databricks hops.
4. Impact analysis — From orders, ask “what breaks if I rename total_amount?” Walk downstream topics and gold aggregates.
5. Classification / PII — customers.email and customers.phone are good targets for auto-classification or manual PII.Sensitive tags.
6. Profiler & data quality — Add a profiler workflow on Postgres or Databricks tables; create a uniqueness test on customers.email and a range test on order_items.quantity.
7. Glossary & ownership — Create a glossary term “Lifetime Order” and link it to orders + gold.customer_order_summary; assign an owner team.
8. Pipelines — Open the Kafka Connect service to see Debezium / sink connectors as pipeline entities with lineage into topics.
9. Live CDC — make live-data, wait a minute, re-run silver/gold SQL, re-ingest, and watch row counts / lineage freshness move.

6. OpenMetadata as an AI context layer (MCP)

OpenMetadata 1.13 ships a built-in MCP server. AI assistants (Cursor, Claude, ChatGPT) can call tools against your live metadata graph instead of guessing about tables.

Enable / connect

1. In the OpenMetadata UI, open Settings → MCP (or follow OpenMetadata MCP docs) and create an MCP connection / token if prompted.
2. Point your MCP client at the server (typical local URL shape):
   • HTTP: http://localhost:8585/mcp
   • Authenticate with the same JWT / SSO your deployment uses.
3. In Cursor: add an MCP server entry that targets that URL, restart the agent.

Example prompts for agents

• “What is the upstream lineage of demo.gold.customer_order_summary?”
• “Which Postgres columns feed demo.bronze.orders?”
• “Who owns the customers table, and is email tagged as PII?”
• “If products.unit_price changes type, what gold tables are impacted?”
• “Create a glossary term ‘Lifetime LTV’ and link it to demo.gold.customer_order_summary.”
• “Draft a data quality test for non-null orders.customer_id.”

Useful MCP tools (names may vary slightly by version): search_metadata, get_entity_details, get_entity_lineage, create_lineage, create_glossary_term, create_test_case, root_cause_analysis.

This is the point of the demo for AI: the agent grounds answers in governed metadata and lineage, not in free-text guesswork about your lakehouse.

7. Troubleshooting

Symptom	Fix
Connect never becomes healthy	docker compose logs connect — first build downloads connector zips; needs network
Debezium fails with replication slot	Ensure Postgres started with wal_level=logical (compose sets this). docker compose down -v and recreate if the volume was initialized without it
Databricks sink UNAUTHORIZED on S3	Check bucket policy on the IAM user keys in .env
Databricks COPY INTO targets default`.`order_items	Sink must set delta.lake.catalog/delta.lake.database (demo/bronze). Re-run make connectors after updating .env / template; create demo.bronze.* first
Databricks COPY INTO DELTA_MERGE_INCOMPATIBLE_DATATYPE (Long vs Int)	Recreate bronze with INT ids via databricks/recreate_bronze.sql, then make connectors
Databricks COPY INTO DELTA_FAILED_TO_MERGE_FIELDS on placed_at / timestamps	Postgres TIMESTAMPTZ is Avro STRING (io.debezium.time.ZonedTimestamp) — recreate bronze with STRING temporal columns (recreate_bronze.sql), cast with to_timestamp() in silver, then make connectors
Databricks COPY INTO COPY_INTO_SCHEMA_MISMATCH (missing line_total/partition, extra __deleted)	Run databricks/recreate_bronze.sql then make connectors — bronze must match Debezium unwrap Avro
Databricks COPY INTO / sink commit fails with S3 403	Storage credential + external location missing, External ID wrong, or location does not cover s3://$S3_STAGING_BUCKET/
Databricks sink ClassNotFoundException: com.simba.spark.jdbc.Driver	Rebuild Connect image (docker compose build connect) — the image includes a JDBC URL shim
Unity Catalog ingestion fails	Confirm token, host, HTTP path; warehouse must be running
No topic→bronze edges	Run make ingest then make lineage
Step 7 ModuleNotFoundError: openmetadata_rest	Fixed for OM 1.13 — pull latest ingestion/add_lineage.py (uses OpenMetadataConnection)
Step 7 SKIPs all bronze tables	UC catalog exists but tables are not ingested yet: run databricks/setup_databricks.sql, confirm sink/COPY INTO, then make ingest && make lineage
OpenMetadata migrate fails	docker compose logs migrate; verify openmetadata_user exists in Postgres

8. Caveats

• The Confluent Databricks Delta Lake Sink is a premium connector with a 30-day trial license when self-hosted. It is append-only — bronze is an event log; silver CTAS dedupes to current state (a realistic medallion pattern).
• Your machine needs outbound access to Databricks and S3.
• Databricks Free Edition is non-commercial, rate-limited, no SLA; inactive accounts may be deleted.
• No always-on Airflow — ingestion is intentionally one-shot to save RAM.

9. Cleanup

make down          # removes containers + volumes
# optionally: delete the S3 staging bucket and IAM user/role when finished