KinetiShift for Azure Databricks

KinetiShift reads changes from your source database (Oracle through LogMiner, or PostgreSQL, MySQL, SQL Server, MongoDB, Kafka) and writes them into Unity Catalog managed Delta tables through the Databricks SQL Statement Execution API, near-real-time (seconds-class). It runs in your own environment; no JDBC or ODBC driver is involved. This guide assumes you deployed KinetiShift from Azure Marketplace (deployment guide); the same settings work on the command line as --flag value.

Supported clouds. Tested end to end on Azure Databricks. KinetiShift uses only the documented SQL Statement Execution API and OAuth, which Databricks also provides on AWS and Google Cloud; there, sign in with a Databricks-managed service principal (OAuth M2M, see Sign-in). We have not yet tested AWS or Google Cloud end to end.

A complete Oracle example

Two Oracle tables, SALES.ORDERS and SALES.ORDER_LINES, copied once and then kept current in main.sales of an Azure Databricks workspace, signing in with the Container App's managed identity.

Pipeline

source = "logminer"
tables = "SALES.ORDERS,SALES.ORDER_LINES"     # or "SALES.*" for the whole schema
poll-interval-secs = 2
initial-load = true                           # copy existing rows first
apply-ddl = true                              # follow ADD / DROP / RENAME COLUMN

sink = "databricks"
databricks-host = "adb-1234567890123456.7.azuredatabricks.net"
databricks-warehouse-id = "abcdef0123456789"
databricks-azure-managed-identity = true
databricks-catalog = "main"
databricks-schema = "sales"
databricks-checkpoint-table = "main.sales._kinetishift_checkpoint"

# optional, see the sections below
databricks-bulk-load-volume = "main.sales.kinetishift_stage"
databricks-column-types = "typed"
databricks-unconstrained-number = "decimal(38,0)"
databricks-column-mapping = true
databricks-table-property = ["delta.enableTypeWidening=true"]
dead-letter-file = "/var/lib/kinetishift/dead-letter.jsonl"

metrics-addr = "0.0.0.0:9100"

The host is the workspace URL without https://. The warehouse id is under SQL Warehouses → your warehouse → Connection details (the last part of the HTTP path). A serverless SQL warehouse is the simplest choice.

Connection environment

KINETISHIFT_ORACLE_CONNECT=//dbhost.example.com:1521/ORCLCDB
KINETISHIFT_LOB_FETCH_CONNECT=//dbhost.example.com:1521/ORCLPDB1
KINETISHIFT_ORACLE_USER=C##KINETISHIFT
KINETISHIFT_ORACLE_PASSWORD=…

KINETISHIFT_ORACLE_CONNECT is where LogMiner runs: the container database root of a multitenant database. KINETISHIFT_LOB_FETCH_CONNECT is the pluggable database that holds the tables; the initial load reads from it. For a non-multitenant database, set both to the same value. The database needs ARCHIVELOG mode and supplemental logging; the user and grants are below. No Databricks secret is needed with the managed identity.

Oracle user and grants

A dedicated common user, as SYSDBA in the container database root. Replace the table names with yours:

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;          -- once per database (needs ARCHIVELOG mode)
CREATE USER C##KINETISHIFT IDENTIFIED BY "…" CONTAINER=ALL;
GRANT CREATE SESSION, SET CONTAINER TO C##KINETISHIFT CONTAINER=ALL;
GRANT LOGMINING TO C##KINETISHIFT CONTAINER=ALL;
GRANT SELECT ANY DICTIONARY, SELECT ANY TRANSACTION TO C##KINETISHIFT CONTAINER=ALL;
GRANT EXECUTE ON SYS.DBMS_LOGMNR TO C##KINETISHIFT CONTAINER=ALL;
GRANT EXECUTE ON SYS.DBMS_LOGMNR_D TO C##KINETISHIFT CONTAINER=ALL;
ALTER USER C##KINETISHIFT SET CONTAINER_DATA=ALL CONTAINER=CURRENT;   -- see the pluggable databases' tables

-- in the pluggable database, for each captured table:
ALTER TABLE SALES.ORDERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
GRANT SELECT, FLASHBACK ON SALES.ORDERS TO C##KINETISHIFT;   -- the initial load reads AS OF a point in time

For a non-multitenant database, use a regular user without C##, CONTAINER=ALL and SET CONTAINER. KinetiShift connects as this user directly; it connects AS SYSDBA only when the user is SYS (or with oracle-sysdba = "yes" for another user granted SYSDBA).

Before the first start, in Databricks

1. Create the schema if it does not exist: CREATE SCHEMA IF NOT EXISTS main.sales.
2. Add the managed identity to the workspace and grant it access (sign-in and grants below).

KinetiShift then creates main.sales.orders, main.sales.order_lines, the checkpoint table and a small _kinetishift_checkpoint_lease table itself, copies the existing rows, and applies each committed change from then on.

The lease table keeps one writer at a time. When you redeploy or restart, Azure briefly runs the new instance next to the old one; the new one waits (another KinetiShift instance holds the single-writer lease) until the old one has stopped, then continues from the checkpoint.

Choose a write mode

Commits that arrive together are written together: one statement per table per batch, then one checkpoint write. A quiet source sees no added delay. databricks-batch-changes caps a batch (default 10000 changes).

Sign-in

Managed identity (recommended on Azure Marketplace)

1. Copy managedIdentityPrincipalId from the deployment's Outputs and look up its application id: az ad sp show --id <principal id> --query appId -o tsv, or search for that object id under Microsoft Entra ID → Enterprise applications with the application type Managed Identities.
2. In the workspace: Settings → Identity and access → Service principals → Add → Microsoft Entra ID managed, with that application id.
3. On the new service principal, turn on the Databricks SQL access entitlement. Without it every statement is refused with HTTP 403.
4. In the pipeline: databricks-azure-managed-identity = true. The identity needs no Azure role assignments.

Microsoft Entra ID service principal with a secret

Add the application to the workspace the same way. Then set databricks-client-id (application id) and databricks-azure-tenant-id in the pipeline, and DATABRICKS_CLIENT_SECRET=… in the connection environment.

Databricks-managed service principal (OAuth machine-to-machine)

Create it under Service principals → Add, then Secrets → Generate secret. Set databricks-client-id and DATABRICKS_CLIENT_SECRET, without a tenant id.

KinetiShift caches the OAuth token, refreshes it before it expires, and never logs it.

Grants

As a workspace or metastore admin, with the application id in backticks:

GRANT USE CATALOG ON CATALOG main TO `<application id>`;
GRANT USE SCHEMA, CREATE TABLE, SELECT, MODIFY ON SCHEMA main.sales TO `<application id>`;
-- only with databricks-bulk-load-volume:
GRANT CREATE VOLUME ON SCHEMA main.sales TO `<application id>`;
--   (or, for a volume you created: GRANT READ VOLUME, WRITE VOLUME ON VOLUME main.sales.kinetishift_stage TO …)

And CAN USE on the SQL warehouse (SQL Warehouses → your warehouse → Permissions). KinetiShift creates and owns its tables; it needs nothing else. Keep the checkpoint table in the same schema.

Because the KinetiShift identity owns the tables, other users, including workspace admins, cannot read them until you grant it. For example, GRANT SELECT ON SCHEMA main.sales TO `analysts`, run by the schema owner, also covers tables created later.

First copy of the data

initial-load = true copies each table's existing rows at a consistent point, then continues with changes from that point, so nothing is missed or applied twice.

• Without a volume, the copy goes through the same MERGE statements as ongoing changes. This suits small tables.
• With databricks-bulk-load-volume, each table's first copy is written as Parquet files to that Unity Catalog volume (created if it does not exist), loaded with one statement per table, and the files are deleted. Use it for any table with more than a few tens of thousands of rows. databricks-bulk-file-mb sets the file size (default 64).

With a checkpoint table, KinetiShift records when the copy finished: an interrupted copy is re-run on restart, and a re-run never duplicates rows.

Tables and column types

Tables are Unity Catalog managed Delta tables, created on the first change. Key columns come first, NOT NULL, with an informational PRIMARY KEY constraint on MERGE tables, and each table carries a comment naming its source table. Optional: Delta table properties (databricks-table-property, a list) and liquid clustering on the key (databricks-cluster-by-key = true). Existing tables are not altered.

Column types. By default every column is STRING and holds the exact source text, so no value is ever rejected or changed. With databricks-column-types = "typed":

Every value is checked against its column before it is written. A value that does not fit exactly, such as a fraction in a DECIMAL(38,0) column, is never rounded; see changes that cannot be written. Tables that already exist keep their column types.

Schema changes

With apply-ddl = true, an added source column is added to the Databricks table, and a new table matched by a SCHEMA.* pattern is created. By default a dropped column is kept but no longer written, and a renamed column is added under its new name with the values copied. With databricks-column-mapping = true, KinetiShift creates MERGE tables with Delta column mapping and drops or renames the column in place instead. A primary-key column is never dropped. For a table created before you turned this on:

ALTER TABLE main.sales.orders SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name');

Timing. LogMiner reads changes with the table's current definition, so changes that KinetiShift has not read yet when a captured table is altered come back undecodable. KinetiShift handles this by writing a LogMiner dictionary into the redo logs when it starts and every 6 hours (logminer-dictionary-build-hours; this uses the EXECUTE ON DBMS_LOGMNR_D grant above). When a change comes back undecodable, it re-reads from the start of that transaction with the dictionary, which follows the table's definition through the ALTER, and carries on; the log shows re-reading from the start of its transaction. LogMiner reads such a dictionary only from archived logs, so the re-read waits until the redo log that holds the change is archived (the log shows waiting for the redo log that holds this change to be archived). On a quiet database the current log can stay open for hours; ALTER SYSTEM SET ARCHIVE_LAG_TARGET = 900 makes Oracle switch logs at least every 15 minutes. Re-reading also needs the archived logs from the last dictionary onward, so keep at least 6 hours of archived logs (more if KinetiShift may be stopped for longer).

KinetiShift stops with an error that names the table and SCN, rather than skip the changes, only when no dictionary from before the change is available: for example the grant is missing, the archived logs were already deleted, or the table was altered while KinetiShift was stopped before its first dictionary was written. To recover, load the tables again: drop the Databricks target tables and the checkpoint table, then restart with initial-load = true. Setting logminer-undecodable = "skip" skips such changes instead, which loses them.

With typed columns and delta.enableTypeWidening=true, a source change that widens a NUMBER widens the Databricks DECIMAL. Column mapping and type widening upgrade the Delta table protocol, so readers outside Databricks need a Delta version that supports them.

Changes that cannot be written

KinetiShift never guesses. A change it cannot write correctly is skipped, logged, and recorded in the dead-letter file (dead-letter-file, one JSON line per change, with the reason), so it can be replayed after the cause is fixed. This covers:

• a value that does not fit its typed column;
• a row of a table without a primary key, unless you set databricks-keyless = "changelog" (that table then lands as an append-only change log);
• a row Databricks itself rejects. The rest of its batch is still written.

Change log and AUTO CDC

With databricks-write = "changelog", each source table lands as <table>__changes:

• Order is by commit SCN, then _seq. Cast _commit_scn to a number before comparing; as text, '10' sorts before '9'.
• A key change is written as two rows with the same SCN and sequence: D with the old key, then U with the new key, so a keyed apply leaves no stale row.
• Delivery is at-least-once. After a crash the last commit may be appended again, identical to the original, so the AUTO CDC result is the same.

Create the pipeline

1. Download a template: auto_cdc_pipeline.py (a list of tables, SCD type 1 or 2 per table) or auto_cdc_pipeline.sql (one table), and upload it to your workspace.
2. Jobs & Pipelines → Create → ETL pipeline: Unity Catalog, serverless, default catalog and schema where the current-state tables should go, source code = the uploaded file.
3. Settings → Configuration. Python: kinetishift.source_catalog, kinetishift.source_schema, and kinetishift.tables, for example [{"table":"orders","keys":["ORDER_ID"]},{"table":"orders","keys":["ORDER_ID"],"scd":2}]. SQL: kinetishift.source, kinetishift.target, kinetishift.keys.
4. Run it triggered on a schedule, or continuous.

The pipeline's owner needs USE CATALOG, USE SCHEMA, CREATE TABLE and SELECT on the schema. APPLY AS TRUNCATE works with SCD type 1 only. The pipeline's serverless compute is billed to your workspace.

Check the result

• The log stream shows pipeline starting, loaded databricks checkpoint after the first run, and, with an initial load, snapshotted table per table and then initial load complete. Ongoing changes are applied quietly; warnings are the lines to watch.
• In Databricks: SELECT * FROM main.sales._kinetishift_checkpoint shows the last applied position, and DESCRIBE HISTORY main.sales.orders shows each write.
• Compare a row count or a recent row in Oracle and Databricks. The status dashboard on port 9100 (internal) shows lag and throughput.

Common problems