Data Ingestion with Lakeflow Connect

Lakeflow Connect is Databricks’ unified data ingestion platform — “simple and efficient connectors to ingest data from local files, popular enterprise applications, databases, cloud storage, message buses, and more” into the Databricks Data Intelligence Platform.

Lakeflow Connect is one of three pillars of the broader Lakeflow product family:

• Connect — efficient ingestion connectors (this lecture)
• Spark Declarative Pipelines (SDP) — accelerated ETL development (the current name for what was previously Delta Live Tables / DLT — see gotcha below)
• Jobs — reliable orchestration for analytics and AI workloads

All three run on Databricks’ “industry leading data processing engine” (Apache Spark + Structured Streaming), sit under Unity Catalog for unified governance, and write to Delta Lake, Parquet, or Iceberg for optimized storage.

• Breadth of sources — on-prem systems, databases, cloud object storage, SaaS applications, and event streams all ingest through one platform.
• Unity Catalog governance — ingested data lands directly into Unity Catalog–governed tables, so access control and lineage are consistent from the moment data arrives.
• Powers downstream consumers — the same ingested data feeds streaming analytics, data science/ML, data sharing, and BI & reporting without separate pipelines per consumer.
• Efficient compute & storage — managed connectors run on serverless compute and use incremental reads/writes to avoid reprocessing unchanged data.

Databricks gives you three broad ways to get data into the platform:

1. Upload Files — upload local files directly to Databricks, upload a file to a volume, or create a table straight from a local file. Good for one-off, small, manual loads.
2. Standard Connectors — connect to cloud object storage, Kafka, and other sources, supporting all three ingestion methods (batch, incremental batch, streaming).
3. Managed Connectors — fully-automated connectors for SaaS applications and databases, with incremental reads/writes built in.

Customizable connectors for cloud object storage and message buses (e.g. Kafka). Require more configuration than managed connectors but offer greater flexibility. Support all three ingestion methods: batch, incremental batch, and streaming.

Fully-automated connectors built on Lakeflow pipelines, with source-specific authentication, CDC (change data capture), edge-case handling, long-term API maintenance, automated retries, and automated schema evolution baked in. Require minimal configuration (UI, APIs, or SDKs).

Per Databricks docs, managed connectors fall into six categories:

• Database connectors (CDC) — e.g. MySQL, PostgreSQL, SQL Server
• Query-based connectors — direct database queries without CDC configuration
• SaaS connectors — e.g. Salesforce, HubSpot, Jira, Workday
• File source connectors — e.g. Google Drive, SharePoint
• Streaming connectors — e.g. RabbitMQ, event streams
• Community connectors — open-source, community-built

On the first run a managed connector ingests all selected data; subsequent runs ingest only what changed — “faster, scalable, and more cost-efficient” than re-pulling everything each time.

• Loads data as batches of rows into Databricks, often on a schedule.
• Traditional batch ingestion re-processes all records every time it runs — no memory of what was already loaded.
• Common techniques: SQL CREATE TABLE AS SELECT (CTAS); Python spark.read.load().
• Best for smaller, one-time or ad hoc datasets where reprocessing everything isn’t costly.

• Ingests (appends) only new data — previously loaded records are skipped automatically.
• Faster and more resource-efficient than full batch since it processes less data per run.
• Common techniques: SQL COPY INTO; Python spark.readStream (Auto Loader with a timed trigger); Declarative Pipelines CREATE OR REFRESH STREAMING TABLE.

• Continuously loads data rows or small batches of rows as they are generated, so they can be queried in near real-time.
• Micro-batch processes small batches at very short, frequent intervals — this is how Structured Streaming implements “continuous” processing under the hood.
• Common techniques: Python spark.readStream (Auto Loader with a continuous trigger); Declarative Pipelines in continuous trigger mode.

Use the comparison table in §3 to reason about trade-offs along these axes:

• Data volume — small/ad hoc → CTAS; thousands of files → COPY INTO; millions+ → Auto Loader.
• Ingestion frequency / latency needs — one-time or infrequent → batch; scheduled incremental → incremental batch; near real-time → streaming (Auto Loader with continuous trigger, or Declarative Pipelines).
• Idempotency requirements — if re-running a job must not duplicate data, avoid CTAS/spark.read alone; prefer COPY INTO or Auto Loader.
• Schema volatility — frequently changing source schemas favor Auto Loader’s automatic schema evolution over manual CTAS handling.
• Source type & governance needs — SaaS apps/databases with governance and minimal ops overhead → Managed Connectors; cloud storage/Kafka needing custom control → Standard Connectors.

Delta Lake is the open-source storage layer underlying Databricks’ lakehouse. Delta tables hold all three data quality levels of the Medallion Architecture — raw (bronze), cleaned (silver), and aggregated (gold) — as one consistent format, so you don’t need separate storage systems per stage.

• ACID transactions — safe concurrent reads/writes, no partial or corrupted data.
• Time travel — query or restore previous versions of a table for auditing and rollback.
• Unified batch + streaming — the same Delta table can be a sink for both batch and Structured Streaming writes/reads.
• Schema enforcement & evolution — rejects mismatched writes by default, but can be configured to evolve as new columns appear.

1. Ingest Data — data enters Delta Lake via batch, streaming, or both; this is the starting point for all processing.
2. Process & Improve Data Quality — data is incrementally refined as it moves through layers, each stage improving structure, quality, and usability.
3. Bronze Layer (Raw Data) — stores raw, unprocessed data from multiple sources; the foundation for all downstream processing.
4. Silver Layer (Cleaned Data) — data is cleaned, transformed, and enriched; produces structured, analysis-ready datasets.
5. Gold Layer (Business-Ready Data) — curated, aggregated data optimized for reporting, BI, and advanced analytics.

Unity Catalog metadata for every database object is organized in a 3-level namespace: catalog.schema.object (e.g. table, view, volume, function, model).

• Catalog — top-level container, often mirroring an org unit or environment (e.g. prod vs dev).
• Schema — sits inside a catalog; holds tables, views, volumes, models, and functions.
• Object — the actual data asset (table, view, volume, etc.).

You can set a default catalog and schema for a session so you don’t have to fully qualify every object reference (e.g. write my_table instead of my_catalog.my_schema.my_table).

A schema is a container for several distinct object types, each governed the same way by Unity Catalog but holding a different kind of asset:

Catalog
  └── Schema
        ├── Tables     → structured data
        ├── Views      → saved query, no storage of its own
        ├── Volumes    → arbitrary files
        ├── Functions  → registered UDFs
        └── Models     → registered ML model artifacts

• current_catalog() / current_schema() — return the catalog/schema currently active for the session, as strings.
• USE CATALOG / USE SCHEMA — set the session’s default catalog/schema so unqualified object references resolve against it (see §7). Switching catalog resets the current schema to default.

SHOW <object_type> lists objects of that type within the current (or a specified) catalog/schema — same family as SHOW SCHEMAS, SHOW CATALOGS, SHOW FUNCTIONS, etc. Full command table is in §7.

• information_schema.columns — a SQL-standard system view that describes every column of every table/view you have access to: which schema and table it belongs to, its column name, data type, and more. Instead of eyeballing DESCRIBE TABLE one table at a time, you can query it like any other table.
• This query uses the same UNION ALL comparison trick as §8’s VERSION AS OF example — running two filtered COUNT(*)s side by side, one per table, to compare column counts between new_hires_ctas and employees at a glance (handy for sanity-checking that two tables you expect to line up actually do).
• WHERE table_schema = '...' AND table_name = '...' — filters the metadata down to one specific table, since information_schema.columns otherwise lists columns for every table/view you can see.

• LIST '<path>' — lists files stored at a Unity Catalog volume path.
• read_files(path [, option => value, ...]) — a table-valued function that reads files directly into a queryable table shape. It auto-detects the file format (JSON, CSV, XML, TEXT, BINARYFILE, PARQUET, AVRO, ORC) and infers a unified schema across files when no format is specified.

• Spark SQL lets you reference files directly as a table by prefixing the path with a format name: <format>.`<path>` — backticks around the path, not quotes. text.`<path>` reads the file(s) at that path using the text data source: one row per line, in a single STRING column named value, with zero parsing.
• This is a quick way to eyeball raw file content before committing to a parsing strategy — check for a header row, confirm the shape looks like JSON/CSV, spot base64-encoded fields — the same reconnaissance step read_files (above) skips straight past.
• LIMIT 5 caps it to the first 5 lines, same as any other query.

This is CREATE TABLE AS SELECT (CTAS) — the exact batch-ingestion technique named in §3’s comparison table. It creates table_name with a schema inferred from the query, and populates it in one step from read_files(...).

• Same CTAS pattern as above, but instead of SELECT *, it projects only specific columns (ID, FirstName, Country, Role) — read_files can be queried like any other table source, so normal column selection, filtering, joins, etc. all work on top of it.
• format => 'csv' — tells read_files explicitly to parse the source as CSV, instead of relying on auto-detection.
• header => true — treats the first row of the file as column headers rather than data.
• inferSchema => true — asks Spark to infer each column’s data type from the file contents, instead of reading every column in as a string.
• These are passed as key => value options directly inside the read_files(...) call — the general pattern is read_files(path, option_key => option_value, ...), and you can stack as many as you need.
• The explicit column list here also has a second purpose: read_files automatically adds a _rescued_data column to its output (see gotcha below) — selecting the four named columns instead of * is what excludes it from the final table.

• The whole point is to rescue, not drop: instead of silently discarding data that doesn’t fit the schema (or erroring out the whole ingest), the offending value(s) get preserved in this one column so nothing is lost.
• It’s stored as a JSON-formatted string — per the official docs, “a JSON blob with the rescued columns and the source file path of the record.” So it’s a STRING column holding a JSON object, not a native struct — you’d use a JSON function (e.g. get_json_object, or from_json with a schema) to pull specific values back out of it if needed.
• Specifically, the docs name three triggers for a field landing in _rescued_data: it’s absent from the provided/inferred schema, its value’s data type doesn’t match the schema, or its column name has a case mismatch against the schema’s field names.
• A row with no mismatches has an empty/no rescued content for that row — matching the general “only populated when something didn’t fit” design of the feature (see the syntax note below on confirming the exact NULL wording).

• _metadata — a hidden STRUCT column available on any file-based data source (CSV, JSON, Parquet, etc. — whether read via read_files, plain spark.read/DataFrameReader, Auto Loader, or COPY INTO). Like _rescued_data, it isn’t shown by SELECT * unless you ask for it by name — it has to be explicitly selected to appear.
• Fields inside the struct: file_path (STRING), file_name (STRING, with extension), file_size (LONG, bytes), file_modification_time (TIMESTAMP), file_block_start (LONG) and file_block_length (LONG) — the last two describe which byte-range of the file was read.
• This is exactly the “additional context, lineage information, and data governance” use case: instead of just the data, you can carry along where each row came from — which source file, when it was last modified, how big it was — as ordinary queryable columns on the target table, e.g. for auditing which upstream file introduced a given row.
• Dot-notation (_metadata.file_name) pulls out one field from the struct, and — as in the second query above — you can alias it to a friendlier column name that actually lands in your table.

• CREATE OR REFRESH STREAMING TABLE catalog.schema.table AS SELECT ... — the third way to define a table’s contents, alongside CTAS and the explicit-schema form above. Declares a streaming table: a Delta table designed to incrementally process a growing input, handling each source row only once. This is the SQL/Declarative Pipelines side of Auto Loader named in §3/§4’s comparison table.
• STREAM read_files('<dir_path>', format => '<file_type>') — wraps the same read_files table-valued function from earlier in §8 in a STREAM modifier, turning a normal batch read into an incremental streaming source that only picks up new files dropped at the path.
• SCHEDULE EVERY 1 HOUR — tells the pipeline to refresh (check for and ingest new files) once every hour, rather than running continuously. Valid units are HOUR/HOURS (1–72), DAY/DAYS (1–31), and WEEK/WEEKS (1–8) — there’s also a CRON form for more precise schedules. Omitting SCHEDULE entirely runs the table continuously instead of on a fixed interval.
• CREATE OR REFRESH — combines “create if it doesn’t exist” and “update the definition if it does” into one statement, so re-running this same DDL after editing it (e.g. changing the schedule or source path) updates the streaming table in place.

• A second flavor of CREATE TABLE, alongside the CTAS form above: instead of AS SELECT ... FROM read_files(...), you spell out a table_specification — (col_name col_type, col_name col_type, ...) — directly. This creates an empty Delta table with the given schema and zero rows, ready to be loaded separately afterward.
• IF NOT EXISTS — same semantics as the CTAS version: no error if the table already exists. It also won’t alter the schema of an existing table if one is already there with a different shape.
• No USING clause is given, so the table defaults to Delta format — per the official docs, “If USING is omitted, the default is DELTA.”
• INT, STRING — standard Databricks SQL column types (there’s also BIGINT, DOUBLE, BOOLEAN, DATE, TIMESTAMP, and more).

• INSERT INTO — appends new rows, either as literal VALUES or from the result of a SELECT query. Supports Delta schema enforcement/evolution, and a BY NAME option to match columns by name instead of position.
• UPDATE ... SET ... WHERE — modifies existing rows in place. The WHERE clause is optional — omit it and every row gets updated.
• DELETE FROM ... WHERE — removes rows that match the condition. Also optional — omit WHERE and it deletes every row in the table.

A single, atomic operation that applies updates, inserts, and deletes from a source into a target Delta table in one pass — supporting schema enforcement or schema evolution depending on how it’s configured. This makes it a natural fit for slowly changing dimensions (SCD), incremental loads, and change data capture (CDC) scenarios, where a batch of upstream changes (new rows, updated rows, deleted rows) needs to be applied to a target table together.

• MERGE INTO target USING source ON <merge_condition> — compares a source table/view/query against a target Delta table row-by-row using the merge condition (usually a key match like target.id = source.id), then applies one or more WHEN clauses depending on whether each row matched.
• WHEN MATCHED — the merge condition found a row in both target and source. Action is either UPDATE SET ... (update the matched target row) or DELETE (remove it). UPDATE SET * updates every column by name from the source; UPDATE SET col = expr, ... updates specific columns.
• WHEN NOT MATCHED [BY TARGET] — the source row has no match in the target (i.e. it’s new). Action is INSERT — either INSERT * (insert all source columns by name) or INSERT (col1, ...) VALUES (...) for specific columns.
• WHEN NOT MATCHED BY SOURCE — the reverse: a target row has no match in the source (i.e. it’s missing from the incoming data). Action is UPDATE SET ... or DELETE — useful for marking/removing target rows that no longer exist upstream.
• Multiple WHEN clauses can be chained (as in the example above — a conditional delete, then a general update, then an insert), each optionally narrowed further with AND <condition>. Clauses are evaluated in order, and only the first matching clause per row is applied.

• target_table starts with two users (peter, zebi), both status = 'current'. source_table carries three rows describing what should happen to each: peter → 'delete', zebi → 'update' (new email), samarth → 'new' (doesn’t exist in target yet).
• The merge condition ON target.id = source.id matches peter and zebi (both exist in both tables) but not samarth (only in source) — that’s what makes samarth fall through to the WHEN NOT MATCHED clause instead of a WHEN MATCHED one.
• Two separate WHEN MATCHED clauses, each narrowed with AND source.status = ..., route matched rows to different actions based on the value in the source row — not just whether a match exists. zebi (status 'update') hits the first clause and gets its email/status updated; peter (status 'delete') hits the second clause and is removed entirely.
• Result: target_table ends up with zebi (updated email, status update) and samarth (newly inserted, status new) — peter is gone. All three outcomes (update, delete, insert) applied from one source table in a single atomic statement.

DESCRIBE HISTORY table_name returns provenance information for every write to a Delta table — version number, timestamp, the user who ran it, and the operation (e.g. WRITE, UPDATE, DELETE, MERGE). This is the audit trail underpinning Delta Lake’s time travel feature (§5) — you can query or restore any prior version shown here.

• VERSION AS OF <n> — queries the table exactly as it looked at version n (the version numbers come straight out of DESCRIBE HISTORY).
• TIMESTAMP AS OF '<timestamp>' — same idea, but pinned to a point in time instead of a version number.
• Because it’s just a modifier on the table reference, you can use it anywhere a table can appear in a query — including a UNION ALL like the example above, which is a handy way to eyeball what changed (e.g. row counts, or specific rows) between “now” and an earlier version, without altering the table itself.

• spark.sql("...") — runs a SQL string from Python and returns the result as a DataFrame, the PySpark equivalent of a %sql cell (§8). Any SQL from §8 can be run this way inside a Python notebook.
• display(...) — the Databricks notebook function for rendering a DataFrame as a scrollable, interactive table in the cell output, with a built-in option to turn it into a chart. It’s a Databricks notebook feature, not a plain PySpark/Python function — different from a DataFrame’s own .show() (plain text output) or Python’s print().
• try / except Exception as e — wraps the query so a missing table doesn’t crash the whole cell/notebook run with an ugly stack trace.
• if "TABLE_OR_VIEW_NOT_FOUND" in str(e): — checks the exception’s text for this specific Databricks/Spark error condition name, so the except block reacts only to “table doesn’t exist yet” and not to some other, unrelated failure.
• else: raise e — re-raises anything that isn’t the expected missing-table error, so a real bug (bad SQL, permissions issue, etc.) still surfaces loudly instead of being silently swallowed.

• spark.read.format("parquet").load(path) — the plain PySpark DataFrameReader, reading a file straight into a DataFrame. This is Batch ingestion (§3/§4), same category as CTAS — just the Python API instead of SQL, and without going through read_files.
• .write.mode("overwrite").saveAsTable(name) — the DataFrameWriter counterpart: writes the DataFrame out as a managed Delta table (Delta is the default format, as in §8). Together, steps 1–2 are the Python/DataFrameReader equivalent of a single SQL CREATE TABLE ... AS SELECT ... statement — read, then materialize as a table — just split into two explicit calls instead of one.
• mode("overwrite") — replaces the table’s contents (and, per the docs, its schema too — an overwrite isn’t required to match the existing table’s columns) if it already exists, rather than erroring or appending. This is more permissive than CTAS’s IF NOT EXISTS, which refuses to touch an existing table at all.
• f"<catalog_path>.{DA.schema_name}.<file_name>" — an f-string building the full three-level Unity Catalog name (§7) from a literal catalog, a variable schema name (DA.schema_name, from the Databricks Academy lab helper), and a literal table name.
• spark.table(name) — the DataFrameReader shortcut for reading an existing table by name into a DataFrame (equivalent to spark.sql("SELECT * FROM name"), just without writing SQL).

• COPY INTO target_table FROM <location> FILEFORMAT = <format> [FORMAT_OPTIONS(...)] — a SQL command that loads new files from a source location into an existing Delta table. This is the Incremental Batch ingestion technique named in §3/§4’s comparison table, and it’s the table-loading step that pairs with the empty table created just above.
• spark.sql("""...""") — same pattern as this section’s first entry: runs a SQL string from Python. Triple quotes let the multi-line COPY INTO statement be written across several lines without escaping newlines.
• result = spark.sql(...) — COPY INTO returns a DataFrame summarizing the operation (e.g. counts of rows/files inserted vs. skipped) when its result is captured like this, rather than left unassigned.
• result.display() — renders that summary DataFrame as an interactive table, same display() covered above.
• FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true') — same idea as read_files’s header/inferSchema options in §8: treat row 1 of each CSV as column headers, and infer each column’s data type from file contents instead of reading everything as a string.

A worked example of the mergeSchema case above: if files landing at <file_path> start including columns that bronze_table doesn’t have yet, COPY_OPTIONS ('mergeSchema' = 'true') lets the table’s schema evolve to pick up those new columns automatically, instead of the load failing or silently dropping the extra data. Without it, a schema mismatch between source files and target table causes COPY INTO to error.

• spark.readStream.format("cloudFiles") — the Python entry point for Auto Loader, the third ingestion technique from §3/§4. "cloudFiles" is the special Structured Streaming source name that activates Auto Loader’s incremental file-detection engine, instead of a normal one-time batch read.
• .option("cloudFiles.format", "json") — tells Auto Loader what format the source files are in (also supports CSV, Parquet, Avro, text, binaryFile, ORC — same format family as read_files in §8).
• .option("cloudFiles.schemaLocation", ...) — a path where Auto Loader persists the schema it infers, so it remembers the schema (and any evolution) across restarts instead of re-inferring from scratch every run.
• .load("/Volumes/catalog/schema/files") — the source directory Auto Loader watches for new files.
• .writeStream.option("checkpointLocation", ...) — a separate path (conceptually distinct from schemaLocation, though both are often placed under the same parent directory) where Structured Streaming tracks processing progress/state, enabling exactly-once, resume-safe streaming.
• .trigger(processingTime="5 seconds") — runs as a micro-batch stream, checking for new files every 5 seconds — this is the “Streaming” row from §3’s comparison (near real-time via frequent small batches), as opposed to a one-off scheduled run.
• .toTable("catalog.database.table") — the three-level Unity Catalog target table (§7) this stream continuously writes into.

• Reduces inferred-schema inconsistency risk — especially for semi-structured sources like JSON or CSV, where Spark has to guess types from sampled data. An explicit schema removes the guessing entirely: every column’s name and type is exactly what you declared, run after run.
• Faster parsing/loading — Spark can apply the declared types and structure immediately, rather than first scanning data to work out what the schema probably is.
• Lower compute overhead at scale — schema inference isn’t free: Auto Loader’s own docs describe sampling “the first 50 GB or 1000 files that it discovers, whichever limit is crossed first” just to infer a schema before real ingestion even starts. On a large or frequently-refreshed source, skipping that sampling pass every time by supplying the schema yourself is a real, measurable savings.
• This is also why the inferSchema/inferColumnTypes naming ambiguity flagged elsewhere in these notes (§8) stops mattering once you provide an explicit schema — there’s no inference step left to worry about.

• schema => 'order_id INT, email STRING, transactions_timestamp BIGINT' — a DDL-format schema string, passed as a single option, telling read_files exactly what columns to expect and what type each one is. No inference happens once this is supplied.
• sep => '|' — the field delimiter option for CSV (default is ,); here the source file is pipe-delimited instead of comma-delimited.
• rescuedDataColumn => '_rescued_data' — names the column that holds anything that doesn’t fit the schema above. Any row with an extra field, a missing field, a type mismatch, or a column-name case mismatch against this schema gets that offending data rescued into _rescued_data instead of the row being dropped or the whole read failing — same mechanic already covered in §8, just now paired with an explicit schema instead of an inferred one.

• The scenario: after an initial ingest (or a check of the raw _rescued_data values), you notice entries keyed like c0/_c0 inside _rescued_data instead of the real column name. That positional-style key (_c0, _c1, …) is Spark’s standard fallback naming for a column it couldn’t map to a name — e.g. the header row didn’t parse the way the schema expected, so the first column’s value got rescued under a generic positional key rather than being matched to order_id.
• _rescued_data:_c0 — the colon (:) operator, Databricks SQL’s JSON path-extraction syntax. It works directly on a STRING column containing valid JSON (exactly what _rescued_data is), pulling out the value at key _c0 without needing a separate from_json parse step.
• CAST(... AS BIGINT) — the colon operator returns the extracted value typed the same as the JSON it came from (effectively untyped/string-like), so it’s cast to the correct target type — here recovering what should have been the order_id column.
• AS order_id, * — the recovered, correctly-typed value is aliased back to its intended column name and selected alongside everything else (*, which still includes the original columns and the still-present _rescued_data struct) — effectively a manual repair of the one column that failed to parse.

• A JSON object (the outer { ... }) is a set of key/value pairs — same idea as a Python dict or a row of named columns.
• A value’s type can be a string ("name": "John Doe"), number ("age": 35), boolean, array ("children": [...]), object ("address": {...}), or null.
• Objects can be flat (every value a simple string/number/boolean) or nested (a value is itself an object or an array of objects) — the address field above is a nested object, and children is an array of nested objects.
• How deep and irregular this nesting gets is exactly what determines how hard the source is to work with downstream — a flat JSON file behaves a lot like a CSV with named columns; deeply nested/variable JSON is where STRUCT and VARIANT (below) start to matter.

• STRING — store the JSON as raw text, untouched. Simplest option, but you can’t query individual fields without a JSON function on every access, and there’s no schema enforcement at all.
• STRUCT — parse the JSON into a fixed, defined schema (via from_json, below), giving you a proper typed, nested column you can dot into directly (e.g. col.address.city). Requires knowing/declaring the schema up front, and a schema change on the source breaks it.
• VARIANT — a semi-structured type built to store any JSON shape flexibly, without a fixed schema, while still being efficiently queryable (path syntax like col:address:city). Per the official docs, “the improved read and write performance for variant allows it to replace native Spark complex types such as structs and arrays in some use cases” — Databricks explicitly recommends VARIANT over plain JSON strings for semi-structured data going forward.

• Instead of hand-writing a DDL schema string (the way §10’s read_files example did), schema_of_json(jsonStr) takes one example JSON-formatted string and automatically derives the matching schema definition from it — useful when the shape is complex or you don’t want to type it out by hand.
• The nested address object becomes a nested STRUCT<...>, and the children array of objects becomes ARRAY<STRUCT<...>> — the returned schema mirrors the JSON’s own nesting exactly, object-for-object and array-for-array.
• The result is returned as a plain STRING — a schema definition, not parsed data — which is exactly the shape from_json (next step) expects as its schema argument.

• from_json(jsonStr, schema [, options]) — takes a STRING column holding JSON text plus a schema (a DDL string, or a schema_of_json(...) call directly), and returns a new column typed as a STRUCT matching that schema.
• This is Step 1 + Step 2 as a pipeline: derive the schema once with schema_of_json on a representative sample, then feed that schema into from_json to actually parse every row’s JSON string into a real typed/nested column — from here on, you can dot into fields directly (e.g. struct_column.address.city) instead of using JSON string functions.
• Column names in the schema must match the JSON’s own key names exactly (case-sensitive) — a mismatch there behaves the same as the schema-mismatch cases already covered for read_files/_rescued_data in §8/§10, just via a different function.

• The walkthrough breaks building a STRUCT schema for nested JSON into five steps: (1) define the schema for the JSON string overall, (2) wrap it in STRUCT<...>, (3) declare simple fields’ types (name: STRING, age: INT), (4) declare a nested-object field as its own STRUCT<...> (address), and (5) declare an array-of-objects field as ARRAY<STRUCT<...>> (children).
• The key distinction highlighted here: a single nested object (address, one city/state pair) is just STRUCT<...>, but a list of similarly-shaped objects (children, multiple name/age pairs) needs the ARRAY<> wrapper around the STRUCT<> — getting this wrong (e.g. declaring children as a plain STRUCT instead of ARRAY<STRUCT>) is a common schema-mismatch source, and would land those rows’ children data in _rescued_data (§8/§10) instead of parsing correctly.

Streaming/event sources (Kafka-style key/value columns, for example) commonly base64-encode their payloads. Base64 turns arbitrary bytes into a safe, printable ASCII string, which avoids corruption or encoding issues in transport (nulls, special characters, binary-safety) — but it also means the value isn’t human-readable until it’s decoded.

unbase64(expr) reverses this: it takes a base64-encoded STRING and returns the decoded value as BINARY — not a string. To get back to human-readable text, that binary result still needs to be CAST to STRING: CAST(unbase64(col) AS STRING).

This is the “bronze layer 2” pattern: bronze_table_raw holds the data exactly as ingested — base64-encoded key/value columns and all — and this step decodes those columns into a new, separate table (decoded_key / decoded_value) without touching or overwriting the raw source. decoded_value is now a JSON-formatted STRING column, ready for the STRING, STRUCT, and VARIANT flattening approaches below.

Once a column holds a JSON-formatted STRING (like decoded_value above), the colon-path operator already introduced in §10 (col:key) can pull out its top-level — and nested — fields directly, with no conversion to STRUCT or VARIANT at all.

• Each decoded_value:<key> extracts one top-level field from the JSON string — same colon-path syntax as §10.
• A colon-path result can itself hold JSON: decoded_value:geo is called out here as potentially another nested JSON-formatted string, and decoded_value:items as a nested array of JSON-formatted strings. Going one level deeper (e.g. decoded_value:geo:city) works the same way.
• This is the “go deeper as needed” version of the STRING approach — no upfront schema, just chain colon-paths further for whatever nested field you need, at the cost of no type enforcement and needing a colon-path (or JSON function) at every level.

Converting the same decoded_value column to a STRUCT follows the two-step pattern already covered above: derive the schema with schema_of_json, then apply it with from_json.

• * EXCEPT (decoded_value) selects every column from bronze_table_decoded except decoded_value itself — shorthand for “keep everything else, I’m replacing this one column.”
• from_json(decoded_value, schema) parses that excluded column into a new, typed value STRUCT column, using the schema string produced by schema_of_json.
• Net effect: same columns as before, minus the raw JSON string, plus a new structured value column you can dot into.

With decoded_value now a STRUCT (value), dot-notation reaches any field directly — value.device for a top-level field, value.geo.city for a nested one — no colon-paths or JSON functions needed anymore. array_size(value.items) counts how many elements are in the items array per row, without unnesting it.

• explode(collection) is a table-valued generator function: it un-nests an ARRAY (or MAP) and returns one row per element, duplicating the other selected columns across those rows. explode(value.items) AS item_in_array turns the 3-element items array shown below into 3 separate rows, each holding one item’s STRUCT.
• A NULL array produces zero rows, not one row of NULLs — use explode_outer() instead if rows with a NULL array need to be kept.
• The un-exploded value.items column is also selected here, so the full 3-element array is visible alongside each individual exploded element — the table below shows the exploded rows.

The VARIANT approach follows the same bronze-layer shape as STRING and STRUCT above, but uses parse_json instead of from_json — and needs no schema at all.

• parse_json(jsonStr) takes a JSON-formatted STRING and returns a VARIANT value representing the same data — no schema argument needed, unlike from_json. If the string isn’t valid JSON, parse_json raises an error (MALFORMED_RECORD_IN_PARSING); use try_parse_json instead to get NULL back for malformed input rather than failing the query.
• This mirrors the STRUCT pipeline exactly — same source table, same decoded column in, one new typed column out — just with a schema-free VARIANT column instead of a fixed STRUCT.

Same colon-path syntax as STRING (and as §10), but on a VARIANT column instead of a STRING one. The difference: a colon-path on VARIANT returns another VARIANT value by default, so the :: shorthand cast (equivalent to CAST(... AS STRING)) is used to get a plain STRING back out — json_variant_value:device::STRING. Left uncast, json_variant_value:items stays a VARIANT holding the nested array.

• Lakeflow Connect Managed Connectors simplify ingestion from enterprise databases and SaaS applications with a fully managed, UI-driven or API-driven experience.
• SaaS ingestion architecture: a serverless Declarative Pipelines job collects credentials from Unity Catalog, reaches out to the public data source, and writes to Streaming Delta Tables.
• Database ingestion architecture: uses an additional Ingestion Gateway (classic compute) to connect to private databases, with staging and state management in a Unity Catalog volume, before a serverless pipeline writes to Streaming Delta Tables.
• Partner Connect provides an alternative when no native managed connector is available, offering a rich ecosystem of partner solutions.

Everything covered so far in this lecture’s code syntax (§8/§9) — CTAS, COPY INTO, and Auto Loader — is built for ingesting from cloud object storage (files in a volume, an S3/ADLS/GCS path). Enterprise databases (on-prem or cloud-hosted MySQL, PostgreSQL, SQL Server, Oracle, etc.) and SaaS applications (Salesforce, Workday, ServiceNow, and similar) are different kinds of sources entirely — they’re not files sitting in object storage, they’re live systems reached over a network connection, each with their own authentication model, API, and change-tracking mechanism. That’s the specific gap Managed Connectors (already introduced conceptually in §2) exist to fill.

• Managed Connectors are built into Databricks — no separate tool or vendor account needed — and can be set up either through the Databricks UI (“point and click”) or the API/SDK for a code-first workflow.
• Per the official docs, they fall into two broad ingestion architectures depending on the source type: SaaS ingestion (for sources reachable over a public API/endpoint) and Database ingestion (for databases, which need continuous change capture). Both are detailed below.
• Not every connector is Generally Available at the same time — the source material’s own note that “managed connectors in Lakeflow Connect are in various release states” lines up with Databricks’ standard release-stage terminology (Private Preview → Beta → Public Preview → General Availability); worth checking a specific connector’s current release stage before relying on it for production.

Per the official docs, the point-and-click path into a managed connector is: in the Databricks workspace sidebar, click Data Ingestion → this opens the Add data page → under Databricks connectors, pick the source you want (e.g. Salesforce, SQL Server). The Add data page is the single landing spot for every ingestion path covered in this lecture, not just managed connectors — it groups them by category:

• Databricks connectors — the Lakeflow Connect managed connectors covered in this section (SaaS apps, databases).
• File upload / local files / cloud object storage — the CTAS/COPY INTO/Auto Loader-style paths from §3, §8, §9.
• Partner connectors — the Partner Connect ingestion partners covered below.

So the “gap” diagram earlier in this section and the Add data page are really the same map, one conceptual and one literal: cloud storage, databases, and enterprise apps all eventually funnel through this one UI entry point, just into different categories of it.

• A SaaS connector’s pipeline is entirely serverless — a Lakeflow (Serverless) Declarative Pipelines job does the whole job: authenticate using credentials stored in Unity Catalog, call out to the source’s public API/endpoint, and write the result to a Streaming Delta Table.
• Per the docs, SaaS connectors use three components: connections (the Unity Catalog securable object holding auth details), ingestion pipelines, and destination tables — and are automatically compatible with serverless egress controls, since there’s no separate always-on infrastructure to manage.
• No dedicated gateway or staging layer is needed here — unlike the database architecture below — because a public API can simply be called on demand each pipeline run.

• Database sources need one extra piece the SaaS architecture doesn’t: the Ingestion Gateway. Per the docs, it runs continuously as its own job on classic compute (not serverless) so it can extract snapshots, change logs, and metadata from the source database in real time — it has to stay running so change logs aren’t truncated at the source before they’re captured.
• The gateway writes what it captures to staging storage — a Unity Catalog volume that temporarily holds the extracted data. This decouples continuous change capture (gateway) from the actual load into Delta tables (pipeline), so the pipeline can run on its own schedule independently of the gateway’s constant capture. Per the docs, staged data auto-purges after 30 days.
• From there, a separate Managed Ingestion job — serverless, same as the SaaS path — reads the staged data and writes/merges it into the destination Streaming Delta Table(s), using CDC to apply just the changes rather than reprocessing everything.
• Officially documented managed database connectors (as of this writing) cover MySQL, PostgreSQL, Microsoft SQL Server, and Oracle, each with a CDC mechanism specific to that database engine.

Partner Connect lets you create trial accounts with select Databricks technology partners and connect your Databricks workspace to partner solutions directly from the Databricks UI — letting you try a partner’s ingestion (or BI, or other) tooling against your own lakehouse data before committing to full adoption. It’s the fallback for sources that don’t have a native Lakeflow Connect managed connector yet.

Beyond the ingestion methods already covered (§2–§4, §8, §9, §12), Databricks offers a handful of additional features for integrating with and sharing data: Lakehouse Federation (query external sources without moving the data), Zerobus (a direct, high-throughput event-ingestion API), Delta Sharing (secure read-only data sharing), and the Databricks Marketplace (an open exchange for data products, built on top of that sharing protocol).

• Lakehouse Federation is Databricks’ query federation platform — it gives governed, read-only access to external data through Unity Catalog foreign catalogs, without copying or moving that data into Databricks first. This is the “data virtualization” framing from the lecture: query it where it lives.
• Queries are pushed down to the external system where possible (via JDBC for relational sources) — so filtering/aggregation happens at the source, and only the result set comes back to Databricks — with Unity Catalog governance (permissions, auditing) still applying on top.
• Supported external sources include MySQL, PostgreSQL, SQL Server, Oracle, Teradata, Amazon Redshift, Snowflake, Google BigQuery, Azure Synapse, Salesforce Data 360, and other Databricks workspaces/instances — plus catalog-level federation for Hive metastore, AWS Glue, and Palantir Foundry.
• Best suited for ad hoc reporting, BI, and proof-of-concept access to operational databases — not a replacement for actually ingesting data you’ll query heavily or repeatedly (the ingestion methods in §2–§4/§12 are still the better fit there).

• Zerobus Ingest is a Lakeflow Connect component for writing event data directly into Unity Catalog Delta tables at high, sustained throughput — skipping any intermediate queuing/streaming system (like standing up your own Kafka cluster) between the producer and the lakehouse.
• It offers multiple interfaces depending on the producer: gRPC-based SDKs for the highest sustained throughput (high-volume streaming producers), a stateless REST API (suited to large fleets of lightweight/“chatty” edge devices), OpenTelemetry support (for systems already emitting traces/logs/metrics), and a Kafka-compatible API (in beta, for migrating existing Kafka producers).
• Supports JSON, Protocol Buffers, and Apache Arrow message formats, with schema validation and durable fallback mechanisms built in.

Delta Sharing is the open protocol for secure, read-only data sharing — it lets you share tables, views, volumes, notebooks, and models with recipients outside your organization, regardless of whether they use Databricks, without copying the data into their system. Real-time: when the provider updates the shared data, recipients see the update in near real time.

Databricks Marketplace is an open exchange where data providers, software vendors, and technology partners publish offerings that customers can discover, evaluate, and connect to directly from their own workspace — browsing and consuming data products without leaving Databricks. It uses OpenSharing (Delta Sharing) under the hood for the actual secure data exchange, and integrates with Partner Connect for pre-configured connections to select technology partners.