Lakeflow Connect, JDBC/REST Clients and JSON Ingestion in Databricks (Data Engineer Associate)

A Lakeflow Connect managed connector is configured from two objects: a Unity Catalog connection, which stores the source's credentials and can be reused, and an ingestion pipeline, which lists the source objects to copy, the destination catalog and schema, per-table settings and the schedule. The pipeline writes streaming tables in that Unity Catalog schema, so the ingested data is governed by catalog grants and lineage from the first run.

You can create the pipeline in the UI wizard, with a Declarative Automation Bundle (formerly Databricks Asset Bundles), the REST API or the CLI. Lakeflow Jobs or the pipeline's own schedule triggers it. External locations govern where managed storage lives, but they do not choose the target tables. Per-table options sit under table_configuration in the pipeline spec:

objects:
  - table:
      source_schema: dbo
      source_table: customers
      destination_catalog: main
      destination_schema: crm_raw
      table_configuration:
        scd_type: SCD_TYPE_2
        exclude_columns: [ssn]

Which connector to choose (managed vs standard connector, query-based vs CDC, federation vs ingestion) is covered in the lesson on ingestion patterns. This lesson covers configuring and running the connector once it is chosen.

For database sources such as SQL Server, the ingestion gateway must run continuously, while the ingestion pipeline can run on a schedule. The gateway runs on classic compute, reads snapshots and change data (change data capture or change tracking) from the database and stages them in a Unity Catalog volume. The ingestion pipeline runs on serverless compute and applies the staged changes to the destination streaming tables.

The gateway is continuous because the database keeps its change log only for a retention period. If the gateway is stopped, scheduled or paused for longer than that, changes are removed from the log before they are captured, and the destination silently misses them. Databricks says not to stop the gateway manually. Saving cost by scheduling the gateway therefore trades compute for permanent data gaps.

• Prevent gaps: keep the gateway continuous and put the cheaper, serverless ingestion pipeline on whatever schedule meets the freshness need (hourly, a few times a day, daily).
• Repair gaps already lost: only a full refresh of the affected tables re-reads them from the source.
• SaaS connectors such as Salesforce, Workday or ServiceNow have no gateway; the ingestion pipeline calls the application's API directly and can be scheduled.

By default a managed connector keeps only the current version of each row (SCD type 1): an update at the source overwrites the destination row. Setting scd_type: SCD_TYPE_2 in a table's table_configuration (or History tracking: On in the UI) keeps every version as its own row, with __START_AT and __END_AT columns marking when each version was current.

Delta time travel is not a substitute for SCD type 2: it reaches back only as far as the table's retained files, and it shows whole-table snapshots rather than per-row validity periods.

A full refresh restarts history. Refreshing an SCD type 2 table replaces it: earlier row versions are removed and tracking starts again from the reload. Before refreshing a history table that people rely on, copy it first (for example with a deep clone or CREATE TABLE AS SELECT) or look for a fix that avoids the refresh.

A managed connector ingests all current and future columns of a source table unless you configure column selection. In table_configuration you set either include_columns or exclude_columns, and the two behave differently when the source gains columns.

Use exclude_columns when a sensitive field (for example a bank account number) must never be copied into Databricks while the rest of the table, including future columns, keeps flowing. Access controls applied after ingestion only restrict who can see data that has already landed; they do not keep the field out of the lakehouse.

Newly selected columns and backfill (Salesforce). For the Salesforce connector, Databricks documents that a column added to the selection after the pipeline has started is not backfilled: incremental runs bring its values only for records that change afterwards, and a full refresh of that table loads its values for existing records. Check the specific connector's documentation before assuming the same for other sources.

Schema changes and when a full refresh is needed

Managed database connectors absorb additive schema changes on their own and need a full table refresh for changes they cannot apply incrementally. For SQL Server, Databricks documents this behaviour:

The full refresh is what reloads the table with the current source schema; the destination stays a copy that the connector manages, so fix schema drift through the connector rather than by hand. When the name clash case arises on a very large table, dropping the inactive column avoids a full reload.

For Salesforce, soft deletes (records in the recycle bin) are applied at the next sync. Records that are hard-deleted, or soft-deleted and purged from the recycle bin before a run, are invisible to incremental reads and stay in the destination until a full refresh. No incremental setting changes that, however often the pipeline runs, because every incremental run reads only what the source still reports as changed.

To pull a database that no managed connector covers, a notebook reads it with Spark's JDBC data source and writes the DataFrame into a Unity Catalog table. ODBC is the protocol most external tools use to connect into Databricks; inside a notebook the Spark JDBC reader is the usual client.

user = dbutils.secrets.get(scope="erp-db", key="user")
password = dbutils.secrets.get(scope="erp-db", key="password")

df = (spark.read.format("jdbc")
  .option("url", "jdbc:postgresql://erp-host:5432/logistics")
  .option("dbtable", "public.shipments")
  .option("user", user)
  .option("password", password)
  .option("partitionColumn", "shipment_id")
  .option("lowerBound", "1")
  .option("upperBound", "25000000")
  .option("numPartitions", "16")
  .option("fetchsize", "10000")
  .load())

df.write.mode("overwrite").saveAsTable("logistics.raw.shipments")

Credentials. Never put a username or password in the JDBC URL, a widget default, a job parameter or a plain-text Spark property. Keep them in a secret scope and read them with dbutils.secrets.get; Databricks redacts secret values printed in notebook output. Secret scopes have their own access control lists with READ, WRITE and MANAGE permissions. A scheduled job runs as its run-as identity, so grant that service principal READ on the scope, the least privilege that lets it fetch secrets; secret access is granted on the scope itself, not through other Databricks permissions.

Parallel reads. Without partitioning options, Spark issues one query over one connection, so a single task reads the whole table while other executors sit idle. Setting all four of partitionColumn, lowerBound, upperBound and numPartitions makes Spark run numPartitions range queries in parallel. Choose an evenly distributed, ideally indexed numeric, date or timestamp column, and keep numPartitions modest so the source database is not overwhelmed. fetchsize tunes rows per round trip; it does not add connections, and nothing applied after the read can parallelize a single query.

The bounds do not filter rows. Spark uses lowerBound and upperBound only to compute the stride of the generated predicates. The first partition also takes everything below the first boundary (and nulls), and the last partition takes everything at or above its start, so every row is read. Bounds far below the real maximum put most rows into the last partition (one huge task). To load a subset, filter in a pushdown query; to balance the read, set the bounds to the column's real minimum and maximum.

Where to write. saveAsTable("catalog.schema.table") lands the data in a named Unity Catalog table, so grants and lineage apply; append adds rows and overwrite replaces them. save() to a path writes files rather than a table: a /Volumes/… path is governed file storage, and a dbfs:/mnt/… mount is legacy storage outside Unity Catalog's table governance.

An incremental JDBC or API pull transfers only rows changed since the last successful load, by filtering at the source on a stored high-watermark and writing the changes with MERGE. Both halves matter: a full read followed by a MERGE still pays to transfer the whole table, and an incremental read followed by an append duplicates rows that were updated in place.

last_ts = spark.sql(
  "SELECT max_updated_at FROM ops.control.watermarks WHERE source = 'crm_customers'"
).first()[0]

changes = (spark.read.format("jdbc")
  .option("url", jdbc_url)
  .option("query", f"SELECT * FROM customers WHERE updated_at > '{last_ts}'")
  .option("user", user).option("password", password)
  .load())

• Push the filter to the source with the query option or a subquery in dbtable, so the database returns only changed rows.
• MERGE on the business key (WHEN MATCHED THEN UPDATE, WHEN NOT MATCHED THEN INSERT) so updated rows replace their old version and new rows are inserted. A retried batch then does not duplicate data.
• Deduplicate change logs first. If the extract is an append-only change log, one key can appear several times in a batch, and Delta Lake fails a MERGE when multiple source rows match the same target row. Rank rows per key by the change timestamp (for example row_number() over PARTITION BY customer_id ORDER BY updated_at DESC) and keep the newest. dropDuplicates on the key alone keeps an arbitrary row.
• Keep the watermark durable, for example in a small Delta control table, and advance it only after the write commits. A window computed from the scheduled trigger time leaves gaps when a run is late or fails, advancing the watermark before the write loses data on failure, and notebook widgets or variables do not persist between runs.

A REST client in a notebook, such as the Python requests library, runs in the driver's Python process, so it does not use the cluster's workers. A loop of API calls executes serially on the driver, the Spark UI shows no tasks, and adding workers or enabling autoscaling changes nothing.

Land raw responses in a volume. A common pattern writes each raw response (JSON pages, CSV extracts) as a file under a Unity Catalog volume, /Volumes/<catalog>/<schema>/<volume>/…, and lets Auto Loader load new files incrementally into a bronze table. Volumes are governed by READ VOLUME and WRITE VOLUME grants, and the raw copy allows replays after a parsing bug. Storage outside Unity Catalog governance either disappears with the cluster (the driver's local disk) or bypasses catalog grants (legacy DBFS locations). Parsing straight into the table is fine when replays are not needed, but it keeps no raw copy.

Orchestrate with Lakeflow Jobs. Client-based ingestion notebooks are usually run as Lakeflow Jobs tasks: a time-based schedule runs them (hourly, daily), task retries re-run a failed task after a delay (useful for transient network or rate-limit errors), and job-level notifications are not sent while failed tasks are being retried, so the failure email arrives only when retries are exhausted (task-level notifications fire on each failed attempt unless Mute notifications until the last retry is selected). Scheduling and retry logic belong in the job configuration, not in notebook code. The run-as identity of the job needs access to the secrets and tables the notebook uses.

Spark reads JSON as JSON Lines by default: one complete record per line. This applies to spark.read.json, read_files and Auto Loader. A pretty-printed document spread over many lines makes every line invalid: with spark.read.json in the default PERMISSIVE mode those lines end up in a _corrupt_record column, while Auto Loader and read_files capture unparsed data in a rescued data column instead. Setting multiLine to true parses each file as whole JSON values; parse-mode and inference settings do not fix the framing.

These techniques apply once semi-structured or unstructured data has landed in a volume or a bronze table, whether a REST client, a file drop or another pipeline put it there; a managed connector's own output tables are configured with the pipeline settings above.

When JSON arrives as a string column (an API body stored in bronze), there are two ways to work with it:

• Typed columns: from_json(col, schema) parses the string with an explicit schema (a StructType or DDL string; schema_of_json can derive one from a sample) into a struct, and select("parsed.*") expands it into top-level typed columns. A string cannot be cast to a struct, to_json goes the other way (struct to string), and explode works on arrays and maps.
• VARIANT: keep the whole document flexible, as described next.

from pyspark.sql.functions import from_json, col

schema = "id INT, status STRING, total DOUBLE"
silver = (spark.table("bronze.orders_raw")
  .withColumn("parsed", from_json(col("body"), schema))
  .select("parsed.*"))

VARIANT for changing JSON

VARIANT is the Databricks column type for semi-structured data whose shape changes: it stores JSON-like values in an efficient binary encoding without a fixed schema, and Databricks recommends it over keeping JSON as strings. A STRUCT needs a schema change whenever the shape changes, and a MAP<STRING, STRING> loses nesting and types.

• Creating values: parse_json(str) converts JSON text into VARIANT and raises an error on invalid JSON. try_parse_json(str) returns NULL for invalid input, so one truncated response does not fail the load and bad rows stay easy to find. CAST(str AS VARIANT) stores the text as a VARIANT string value; it does not parse it.
• Path syntax: a colon after the column name selects a top-level field, then dots (or ['key']) for nested fields and [n] for 0-based array elements: payload:device.readings[2].temp. A dot directly after the column name (payload.device) is struct syntax, not VARIANT path syntax.
• Field names are case-sensitive. A path with the wrong case matches nothing and returns NULL, so a query runs but returns empty values.
• Casting: ::type or variant_get(col, '$.path', 'type') raise an error when a value cannot be converted. try_variant_get (or the ?:: operator) returns NULL instead.

SELECT
  payload:device.readings[2].temp::DOUBLE           AS third_temp,
  try_variant_get(payload, '$.battery', 'int')      AS battery_or_null
FROM bronze.telemetry;

Auto Loader into VARIANT has two modes:

Generator functions turn array elements into rows. Two choices decide which one fits: whether each element should become one column or one column per struct field, and whether a parent row with an empty or null array must survive.

Use an _outer variant whenever parent counts must stay correct, for example customers who have no saved addresses. For an ARRAY<STRUCT<…>> that must become separate columns in a single table-valued function while keeping empty arrays, inline_outer combines both behaviours. flatten is different: it merges nested arrays into one array within the same row and produces no new rows.

Unstructured files such as PDFs and images are ingested with the binaryFile format, which reads each file as one row with path, modificationTime, length and content (the raw bytes). With Auto Loader (cloudFiles.format = binaryFile) new files in a volume are loaded incrementally into a Delta table, and a later step extracts text or features from the bytes. The text format, with or without wholeText, decodes content as strings and corrupts binary data.

docs = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "binaryFile")
  .option("pathGlobFilter", "*.pdf")
  .load("/Volumes/main/legal/contracts/"))

pathGlobFilter selects files by name pattern before they are read, so files with other names in the same folder are never opened. Filtering on the path column after the load gives the same rows but reads every file's bytes first. recursiveFileLookup controls subdirectory traversal, not file types. The files themselves can stay in the volume; storing large binaries in Delta is a choice to make only when downstream processing needs them in a table.