Metadata-Version: 2.4
Name: spark-kindling-ext-databricks
Version: 0.14.0rc1
Summary: Databricks Lakeflow adapter for Kindling's SDP declaration engine
Author: SEP Engineering
Author-email: SEP Engineering <engineering@sep.com>
License-Expression: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: spark-kindling>=0.14.0rc1,<0.15
Requires-Dist: spark-kindling-ext-sdp>=0.14.0rc1,<0.15
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/sep/spark-kindling-framework
Project-URL: Documentation, https://github.com/sep/spark-kindling-framework/tree/main/docs
Project-URL: Changelog, https://github.com/sep/spark-kindling-framework/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/sep/spark-kindling-framework/issues
Description-Content-Type: text/markdown

# kindling_ext_databricks

Databricks Lakeflow adapter for Kindling's SDP declaration engine.

Lakeflow "extends and is interoperable with" the OSS `pyspark.pipelines`
API, so this adapter **augments** the `kindling_ext_sdp` core rather than
translating it: the OSS emission (materialized views with comments, table
properties, partitioning, clustering, schema) runs unchanged, and
Databricks-only capabilities are layered on top.

Install it from PyPI with `pip install spark-kindling-ext-databricks`; in a uv project,
`kindling env add spark-kindling-ext-databricks` pins the version that matches the
project's Kindling release.

Selected with:

```python
kindling.initialize(engine="databricks_sdp")
...
kindling.declare_pipeline()
```

— resolved through kindling core's engine-extension mapping
(`kindling_<name>.engine_extension()`); adding this engine required zero
core changes.

## Phase 3 (this package today): expectations

```yaml
datapipes:
  silver.orders:
    engine:
      databricks_sdp:
        expectations:            # violations counted, rows kept (warn)
          valid_order_id: "order_id IS NOT NULL"
        expectations_drop:       # violating rows dropped
          positive_qty: "quantity > 0"
        expectations_fail:       # violation fails the update
          no_future_dates: "order_date <= current_date()"
```

Blocks map to `expect_all` / `expect_all_or_drop` / `expect_all_or_fail`
decorators. Declaring them while targeting OSS fails fast at validation
(capability gating in the core); declaring them here against a runtime
without the Lakeflow decorators fails at declaration with an actionable
error.

`refresh_policy: incremental` is accepted (adapter-tier gated) but emits
nothing yet — incremental MV refresh on Databricks (Enzyme) is engine
behavior rather than a declaration keyword; treated as a documented hint
pending verification against a live workspace.

## Phase 5: SCD declared flows → AUTO CDC

An SCD-tagged output entity (`scd.type` plus the declared-flow tags from
kindling core) is declared as the Lakeflow CDC pattern — a pipeline-scoped
source view (the pipe's DataFrame), a streaming table target, and an AUTO
CDC flow — instead of a materialized view:

| Kindling declaration | AUTO CDC |
|---|---|
| `scd.type: "2"` (the only type core registration accepts) | `stored_as_scd_type` |
| entity `merge_columns` | `keys` |
| `scd.sequence_by` | `sequence_by` (change feed) |
| `scd.source_kind: change_feed` (default) | `create_auto_cdc_flow` |
| `scd.source_kind: snapshot` / `scd.close_on_missing: true` | `create_auto_cdc_from_snapshot_flow` |
| `scd.delete_when` | `apply_as_deletes=expr(...)` |
| `scd.tracked` | `track_history_column_list` |
| (default) sequence column ≠ content | `track_history_except_column_list=[sequence_by]` |

Fail-fast mapping requirements (validated with everything else, all
errors at once): change-feed targets need `scd.sequence_by`
(`scd_sequence_by_required`); `scd.type` must be `2`, matching core registration
(`scd_type_unsupported` — bitemporal is deliberately excluded while in
Beta); keys are required (`scd_keys_required`).

Documented divergences from the runner engine (dual-engine parity
criterion): history columns are `__START_AT`/`__END_AT` (the entity's
runner-shape schema is not passed to the streaming table); snapshot
ordering is ingestion order, not `scd.sequence_by`; out-of-order
change-feed rows are reconciled into history rather than ignored under
the runner's strictly-later rule; the `scd.current_entity_id`
current-view companion stays runner-only (an ordinary declared view is
the SDP-native replacement, tracked separately). Expectations, when
configured, attach to the source view — quality is checked on the
incoming feed.

Change-feed sources are consumed as a **stream**: every input selected by
`driving_entity_ids` is read with `spark.readStream.table()` so the AUTO CDC
flow processes it incrementally — the declarative replacement for
hand-rolled foreachBatch+MERGE. Remaining inputs stay batch reads
(stream-static joins). Snapshot sources keep batch reads: the snapshot API
diffs whole snapshots per update.

## Provider-owned streaming sources

An external provider stream, initially Event Hubs over Kafka, can drive a
normal Kindling ingestion pipe when targeting `engine="databricks_sdp"`:

```python
DataEntities.entity(
    entityid="stream.telemetry",
    name="Telemetry Event Hub",
    tags={
        "provider_type": "eventhub",
        "provider.eventhub.connectionString": "@secret:lakeflow:eh-conn",
        "provider.eventhub.name": "telemetry",
        "provider.transport": "kafka",
        "provider.kafka.includeHeaders": "true",
        "provider.preprocess": "kafka",
    },
    schema=None,
)

@DataPipes.pipe(
    pipeid="bronze.telemetry.ingest",
    input_entity_ids=["stream.telemetry", "ref.devices"],
    output_entity_id="bronze.telemetry",
    driving_entity_ids=["stream.telemetry"],
)
def ingest_telemetry(stream_telemetry, ref_devices):
    ...
```

The shared declaration planner asks the provider for a secret-safe
`StreamingSourceSpec`. Invalid source configuration is reported before
emission without printing connection strings, SAS keys, JAAS values, or
resolved `@secret` values. The Databricks adapter then emits exactly one
`dp.create_streaming_table(...)` target and one `dp.append_flow(...)`. When
Lakeflow evaluates the append-flow function, Kindling resolves the provider,
calls `read_entity_as_stream()` with the declarative-source option, passes
later inputs as static DataFrames, invokes the registered pipe transform, and
returns the streaming DataFrame to Lakeflow.

The output entity's declared `schema` is not forwarded to
`create_streaming_table` pending Lakeflow platform evidence; the target schema
is inferred from the append-flow DataFrame on this path.

Lakeflow owns query startup, checkpoint placement, retries, update scheduling,
and target persistence for this path. Kindling does not call `writeStream`,
does not pass `checkpointLocation`, and does not start a streaming query.

## Incremental reads of external Delta inputs

A normal pipe whose driving input is an ordinary Delta table — one produced
by a separate ingestion pipeline, not by a streaming provider — can be read
incrementally by opting in per pipe:

```yaml
datapipes:
  silver.device_telemetry:
    engine:
      databricks_sdp:
        streaming_inputs: ["bronze.device_telemetry"]
```

The output is then declared as a streaming table plus one append flow, the
named input is read with `spark.readStream.table(...)`, and every other
input stays a batch read — so reference-table joins remain stream-static.
Everything else on this path (expectations, table properties, partitioning,
clustering, the deferred `schema` forwarding) behaves exactly as it does for
a provider-owned streaming source; both shapes reach the same emission.

**The opt-in is never inferred.** An incremental read changes what a pipe
means — append-only, with no reprocessing of revised rows — so eligibility
is the app's decision, not the adapter's. Nothing changes for a Delta input
that is not named here.

It is adapter-tier: `engine="sdp"` reports `capability_not_supported`.
Rejected shapes, all at declaration time:

| Code | Shape |
| --- | --- |
| `streaming_input_not_an_input` | names an entity the pipe does not read |
| `streaming_input_not_driving` | names a non-driving input |
| `streaming_inputs_partial` | does not name every driving input |
| `multiple_streaming_inputs` | names more than one input |
| `streaming_input_not_delta` | names a non-Delta entity |
| `streaming_input_internal` | names a dataset produced in this pipeline |
| `streaming_input_provider_owned` | names a provider-owned stream (already streamed) |
| `streaming_dataset_type_conflict` | output explicitly requests a materialized view |

`streaming_inputs_partial` exists because the append flow streams whatever
`driving_entity_ids` selects: a partial opt-in would silently stream an
input the app never opted in, so the sets must match.

## Temporal chain execution mode

A temporal chain lowers its event strata `<events>__g0..gK` as streaming
tables fed by append flows, so each base-event transform runs over a
Structured Streaming DataFrame. That rejects ordered analytic windows —
`row_number`, `lag`, unbounded forward fill — even though the same transform
runs correctly under the runner engine. Set the mode to `batch` to lower
those strata as materialized views with batch reads instead:

```yaml
kindling:
  lakeflow:
    temporal_mode: batch
```

From a Databricks Asset Bundle, use the canonical SparkConf spelling in the
pipeline's `configuration:` mapping — it is point-looked-up by the selector,
so it also works on serverless and shared-access runtimes that cannot
enumerate Spark configuration:

```yaml
configuration:
  spark.kindling.lakeflow.temporal_mode: "batch"
```

| Component | `streaming` (default, omitted) | `batch` |
| --- | --- | --- |
| `<events>__g0` | Streaming table, one append flow per base declaration | One materialized view, union of the transformed base inputs |
| Base reads | `spark.readStream.table(...)` | `spark.table(...)` |
| `<events>__g1..gK` | Streaming tables and append flows | Materialized view per generation, batch reads of lower strata |
| Empty generation | Empty streaming projection of `__g0` | Empty batch projection of `__g0`, same schema and dependency |
| Episode snapshot, episodes, determinations, higher-order strata, public events | unchanged | unchanged |

### Unpersisted generations

In `batch` mode the numbered strata can be declared as pipeline-scoped
temporary views instead of materialized views, so no `__g*` tables are
created and nothing intermediate is written:

```yaml
kindling:
  lakeflow:
    temporal_mode: batch
    temporal_strata_materialization: view    # default: table
```

Only the numbered strata change. The determinations view, higher-order
boundaries, the episodes Auto CDC target and the public `events` union are
declared identically, so query results are unchanged.

The cost is recomputation. Generation `k` reads every stratum below it, so a
temporary view is re-expanded once per reference rather than read back from
storage: the plan behind `events` grows exponentially in
`kindling.temporal.max_generations`, on the order of `2^K` rescans of the
base stratum at ceiling `K`. That is fine at a ceiling of 1–3 and a serious
problem at the default of 10. You also give up per-generation observability —
there is no `__g*` table to query when working out which generation produced
an event, and no per-stratum metrics in the event log.

`view` requires `temporal_mode: batch`. A streaming stratum is an append-flow
target and a temporary view cannot be one, so the combination fails the
declaration instead of quietly downgrading to batch reads.

### How many strata get declared

`kindling.temporal.max_generations` is a ceiling, not a target. A chain whose
conditions are all registry-declared (`DataConditions.register`) declares
exactly as many numbered strata as those rules reach — a single-generation
app gets `__g0` and `__g1`, nothing else — because both emission and
`validate()`'s name reservation can compute that depth in-process from the
condition registry. The ceiling still applies: a rule set reaching past it
fails the declaration rather than being truncated.

A chain with any table-sourced condition engine keeps the ceiling as its
topology, declaring `__g0..__gK` whether or not each generation holds rules.
Rule rows are data that changes between updates, and the reservation path
runs with no Spark session to read their depth with, so a fixed shape is the
only way emission and reservation can agree. Empty strata there are real
datasets with a flow that filters everything out, so set the ceiling to the
depth you actually use.

Shrinking the depth of a registry chain leaves the highest `__g*` dataset
behind, no longer updated. That is a deploy-time event; drop the stale table
once.

Multi-source fan-in is unchanged: every base declaration lands in the one
stratum-0 dataset, each input keeping its own transform. Base sources still
resolve to external physical names through `EntityNameMapper` in both modes,
so a producer selected in the same pipeline establishes no local dependency
edge — keep those sources in an upstream resource. A batch chain declares no
`readStream` and no append flow at all; a chain *with* episodes still calls
`create_streaming_table` for the snapshot-CDC target only, because that is
what the AUTO CDC FROM SNAPSHOT API requires.

Semantics to plan for:

- Batch strata carry batch-query semantics over the rows available at each
  refresh. A refresh can revise or remove previously produced events, and
  late arrivals can change window results across a whole subject partition.
  They are not append-only event archives and they do not use runner
  watermarks.
- Retain the source history the computation needs. Removing input rows can
  remove derived events and change the episode snapshot; snapshot CDC keeps
  its SCD2 version history but cannot reconstruct source history that was
  never retained.
- Use stable event identity and deterministic window ordering, including a
  tie breaker for equal timestamps. Never use a mutable row rank as event
  identity.
- Changing the mode changes `__g0..gK` dataset types. It is not a hot toggle
  with portable checkpoint state: select the mode for newly provisioned
  pipeline outputs, and treat conversion of an existing pipeline as a
  deployment operation. Nothing is dropped or reset automatically.
- The value ignores surrounding whitespace and case. Anything other than
  `streaming` or `batch` fails the declaration before any Lakeflow object is
  created. The setting is inert when no temporal chain-events pipe is
  selected, and never changes general SDP pipe execution, Lakeflow triggered
  versus continuous scheduling, or the runner engine's watermark behavior.

Batch mode enables batch analysis; it does not guarantee that arbitrary user
code is declarable, or that a query refreshes incrementally.


## Deferred

- Multiple provider-owned streaming inputs and stream-stream joins.
- Bitemporal AUTO CDC (`stored_as_scd_type="bitemporal"`) — Beta;
  tracked, not adopted (proposal decision).
- Current-view companion as a declared view.


## Output dataset naming

Lakeflow uses the shared [SDP dataset naming configuration](https://github.com/sep/spark-kindling-framework/blob/main/packages/extensions/kindling_ext_sdp/README.md#output-dataset-naming):

```yaml
kindling:
  sdp:
    dataset_naming: leaf
```

With a pipeline destination of catalog `dev_silver`, schema `cwmdp`,
`silver.device_telemetry` is declared as `device_telemetry`, producing
`dev_silver.cwmdp.device_telemetry`. Set separate destinations and select
the appropriate pipes for each medallion resource. Same-leaf outputs in
one plan are rejected.

This also controls AUTO CDC targets and temporary sources, temporal
events/episodes, and their generated strata, snapshots, and internal
references. Cross-pipeline external table resolution remains independent,
including temporal reads through `EntityNameMapper` and entity
`provider.table_catalog` overrides. Omitting the setting preserves the
existing `silver_device_telemetry` convention.


When consuming these outputs, explicitly align external entity metadata with
the leaf table, for example
`provider.table_name: dev_silver.cwmdp.device_telemetry`. Otherwise a consumer
configured with catalog and schema still resolves the historical flattened
name. See [external reads and generated-name reservations](https://github.com/sep/spark-kindling-framework/blob/main/packages/extensions/kindling_ext_sdp/README.md#reading-leaf-named-outputs-from-elsewhere)
for the complete example and the temporal source limitation.

## Canonical configuration from a Bundle

Lakeflow uses the same Kindling configuration hierarchy as jobs and notebooks.
Publish ordinary Kindling settings files and pass them through the canonical
bootstrap key `spark.kindling.bootstrap.config_files`, or publish them through
`spark.kindling.bootstrap.artifacts_storage_path` and let bootstrap discover
base, platform, workspace, environment, and app overlays.

The selector does not parse YAML, inspect files, or validate structured
sections. It bridges Lakeflow pipeline configuration into
`kindling.initialize(..., app_name=<selected>, engine="databricks_sdp")` and
sets `declaration_only`; the shared bootstrap/Dynaconf path owns transport,
parsing, validation, precedence, and overlays. Flat bridged pipeline keys still
act as bootstrap overrides and win over settings files.

Use Bundle sync to deploy config to a stable workspace or volume path. This
matches the transport pattern in the
[DAB config promotion guide](https://github.com/sep/spark-kindling-framework/blob/main/docs/guide/dab_config_promotion.md):

```yaml
# databricks.yml
bundle:
  name: telemetry-lakeflow

targets:
  dev:
    workspace:
      host: https://adb-<dev>.azuredatabricks.net
      file_path: /Workspace/Shared/kindling/dev

sync:
  include:
    - config/**
    - data-apps/**

resources:
  pipelines:
    telemetry_bronze:
      name: telemetry-bronze
      catalog: dev_bronze
      target: cwmdp
      configuration:
        "kindling.data_app": telemetry
        "kindling.lakeflow.pipes": bronze.ingest_telemetry
        "spark.kindling.bootstrap.environment": dev
        "spark.kindling.bootstrap.workspace_id": adb-dev
        "spark.kindling.bootstrap.config_files": '["/Workspace/Shared/kindling/dev/config/settings.yaml", "/Workspace/Shared/kindling/dev/config/settings.databricks.yaml", "/Workspace/Shared/kindling/dev/data-apps/telemetry/settings.yaml"]'
    telemetry_silver:
      name: telemetry-silver
      catalog: dev_silver
      target: cwmdp
      configuration:
        "kindling.data_app": telemetry
        "kindling.lakeflow.pipes": silver.build_telemetry,silver.derive_events,silver.derive_episodes
        "spark.kindling.bootstrap.environment": dev
        "spark.kindling.bootstrap.workspace_id": adb-dev
        "spark.kindling.bootstrap.config_files": '["/Workspace/Shared/kindling/dev/config/settings.yaml", "/Workspace/Shared/kindling/dev/config/settings.databricks.yaml", "/Workspace/Shared/kindling/dev/data-apps/telemetry/settings.yaml"]'
```

```yaml
# data-apps/telemetry/settings.yaml
kindling:
  sdp:
    dataset_naming: leaf

dataentities:
  bronze.device_telemetry:
    tags:
      provider.table_name: dev_bronze.cwmdp.device_telemetry
  silver.device_telemetry:
    tags:
      provider.table_name: dev_silver.cwmdp.device_telemetry
  silver.events:
    tags:
      provider.table_name: dev_silver.cwmdp.events
  silver.episodes:
    tags:
      provider.table_name: dev_silver.cwmdp.episodes
```

With `dataset_naming: leaf`, dotted logical IDs stay unchanged in the Kindling
registry while Lakeflow emits single-part output names inside each pipeline
destination:

| Logical ID | Pipeline destination | Physical table |
|---|---|---|
| `bronze.device_telemetry` | `dev_bronze.cwmdp` | `dev_bronze.cwmdp.device_telemetry` |
| `silver.device_telemetry` | `dev_silver.cwmdp` | `dev_silver.cwmdp.device_telemetry` |
| `silver.events` | `dev_silver.cwmdp` | `dev_silver.cwmdp.events` |
| `silver.episodes` | `dev_silver.cwmdp` | `dev_silver.cwmdp.episodes` |

Pipeline-local outputs get those names from `dataset_naming: leaf` plus the
pipeline catalog/schema. External reads of the same tables still use the
ordinary entity resolver, so they need the fully qualified
`provider.table_name` overrides shown in the structured YAML. A catalog/schema
override alone is not enough: it would still resolve the default flattened
table name. Temporal chain base-event sources always resolve externally, so
keep their external table overrides in the same structured YAML.

Workspace files under `/Workspace/...` are documented as driver-readable for
clusters and jobs. Readability from a Lakeflow serverless pipeline has not yet
been confirmed; if that is required, deploy the same YAML to a Unity Catalog
volume path and reference that path in
`spark.kindling.bootstrap.config_files`.

`kindling.lakeflow.config_files`, the comma-separated alias that 0.12.35
deprecated, was removed in 0.12.47 and is now ignored. Bundle configuration
that still sets it loads no files from it: move the paths to
`spark.kindling.bootstrap.config_files` as a JSON array string (or a bare
string for one file).
