# Spark & Delta Lake

minilake stores two kinds of table, and the difference decides everything else:

| | MANAGED | EXTERNAL + DELTA |
|---|---|---|
| Backed by | A real DuckDB table | Real Delta files (parquet + `_delta_log/`) at `storage_location` |
| Read by SQL | Natively | Through DuckDB's `delta_scan()` |
| Written by SQL | Directly, fast | Intercepted and run as a real Spark job — correct, and seconds not milliseconds |
| Visible to Spark | **No** — there are no files | Yes |

MANAGED is the fast path for test data. EXTERNAL is how you get real Delta semantics and
anything Spark should see.

## The loop that does the work

Spark writes the files; minilake's SQL reads them. Both see the same bytes.

**1. Write Delta by path.**

```python
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)

path = "/data/delta/vendas/loja/eventos"
spark.createDataFrame([(1, "a"), (2, "b")], ["id", "nome"]) \
    .write.format("delta").mode("overwrite").save(path)
```

Paths must be under `/data/` — the volume both minilake and the Spark container see.

**2. Register it once**, so SQL can address it by three-part name:

```python
w.tables.create(
    name="eventos", catalog_name="vendas", schema_name="loja",
    table_type=TableType.EXTERNAL,
    data_source_format=DataSourceFormat.DELTA,
    storage_location="/data/delta/vendas/loja/eventos",
)
```

Or in one statement, which is the native Databricks spelling and registers the table for
you:

```sql
CREATE TABLE vendas.loja.eventos (id BIGINT, nome STRING)
USING DELTA LOCATION '/data/delta/vendas/loja/eventos'
```

**3. Query it** — including joined against MANAGED tables:

```sql
SELECT e.nome, p.cliente
FROM vendas.loja.eventos e JOIN vendas.loja.pedidos p ON p.id = e.id
```

### Column types come from the files

`tables.get()` reports the schema in the Delta log, not the one you declared. Register `id`
as `INTEGER` over a log that says `long` and you get back `LONG` — the declaration loses,
because the physical type is what a query will actually return.

## Running Spark

Jobs run `spark-submit` in a sibling container from `apache/spark:3.5.3-scala2.12-java17-python3-ubuntu`.
Through the MCP server, `run_python_script` does the whole cycle in one call; through the
REST API, create a job with a `spark_python_task` and trigger it.

Anything touching Delta needs the jars, which the base image does not ship. Over MCP that
is `delta=True`; over the Jobs API it is a `libraries` entry with the Maven coordinate
`io.delta:delta-spark_2.12:3.2.1`.

Those jars — and the Unity Catalog connector's — are resolved when the minilake image is
built and copied onto the shared data volume on the first job run, so `--packages` resolves
from the local Ivy cache and no job needs Maven Central. Any *other* coordinate you pass
still does.

## Reading by name with Unity Catalog

`spark.table("catalog.schema.table")` resolves against minilake. Spark's
`io.unitycatalog.spark.UCSingleCatalog` connector calls minilake's Unity Catalog endpoints,
gets the `storage_location`, and reads the Delta files directly — the same protocol it
speaks to a real workspace.

```python
UC = "http://localhost:8000"

spark = (
    SparkSession.builder
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    # Both are required: the named catalog resolves three-part names, and spark_catalog
    # must also be UC-backed or Delta refuses the operation.
    .config("spark.sql.catalog.spark_catalog", "io.unitycatalog.spark.UCSingleCatalog")
    .config("spark.sql.catalog.spark_catalog.uri", UC)
    .config("spark.sql.catalog.spark_catalog.token", "")
    .config("spark.sql.catalog.vendas", "io.unitycatalog.spark.UCSingleCatalog")
    .config("spark.sql.catalog.vendas.uri", UC)
    .config("spark.sql.catalog.vendas.token", "")
    .getOrCreate()
)

spark.sql("INSERT INTO vendas.loja.eventos VALUES (3, 'c')")
print(spark.table("vendas.loja.eventos").count())
```

Run it with the connector on the classpath: `packages=["io.unitycatalog:unitycatalog-spark_2.12:0.2.1"]`
alongside `delta=True`.

Use this when the point is that the job's code matches what runs on Databricks. For moving
data in and out, by path is fewer moving parts.

### Limits

- **EXTERNAL Delta tables only.** MANAGED tables have no files for Spark to read.
- **The connector is version-locked.** `unitycatalog-spark_2.12` stops at `0.2.1`
  (Spark 3.5.3, Delta 3.2.1). Later releases require Spark 4 + Scala 2.13, which brings the
  `catalogManaged` Delta feature — and DuckDB's `delta_scan` cannot read those tables at
  all, because their commit log lives partly in the catalog. Staying on this line is what
  keeps the write-with-Spark/read-with-SQL loop working.
- **No credential vending semantics.** minilake returns an empty credential set, which is
  what the reference Unity Catalog server does for a filesystem location. Nothing about
  grants or scoped access is exercised here.

## JupyterLab

minilake runs a JupyterLab of its own, as a child process, reverse-proxied at `/jupyter` and
framed by the UI at `/ui/notebooks`. Two notebooks are seeded into the notebook directory on
first start; both reach the minilake that launched them at `http://localhost:8000`.

`minilake_quickstart.ipynb` registers an EXTERNAL Delta table, writes to it with real Spark,
and reads the same rows back through minilake's SQL API — the same loop as above,
interactively.

`minilake_delta_optimize_vacuum.ipynb` takes that further into Delta operations: fifty
micro-batch appends to manufacture a small-files problem, `OPTIMIZE` (including `ZORDER BY`)
to compact it, and `VACUUM` to reclaim the tombstoned files. Every measurement — file
counts, and a replay of the `_delta_log` itself — runs through minilake's SQL API, so the
effect of each Delta operation is visible from the Databricks side. `OPTIMIZE`, `VACUUM`,
`DESCRIBE HISTORY` and `VERSION AS OF` are Delta protocol operations and run in Spark; they
are not part of minilake's DuckDB SQL dialect, and the notebook says so rather than hiding
it.

**The kernel has no `pyspark`, deliberately** — the image carries no second Spark. Both
notebooks submit their Spark work through the Jobs API, into the same sibling containers a
job task uses, with the `run_python_script` helper from the recipe above. A bare
`SparkSession` in a cell will fail, and `pip install delta` installs an unrelated PyPI
package that will not fix it.

For a kernel that *does* hold a local `SparkSession`, there is an optional profile with
PySpark and `delta-spark` preinstalled, sharing the same data volume:

```bash
docker compose --profile notebook up -d
# http://localhost:8888 — everything under notebooks/ is pre-loaded
```

There it is `minilake_delta_quickstart.ipynb`, and minilake is at `http://minilake:8000`
rather than localhost, because the notebook runs in its own container.