Skip to content

Commit 9bb3d0e

Browse files
Aditya GoyalAditya Goyal
Aditya Goyal
authored and
Aditya Goyal
committed
Databricks: detect Delta table version changes (Phase 1) (#74195)
Add DatabricksDeltaTableVersionSensor and DatabricksDeltaTableVersionTrigger to monitor Delta table commit versions (Phase 1: scheduled-task sensor and deferrable trigger): - Rebase onto #74191 and reuse UnityTableIdentity from airflow.providers.databricks.assets.databricks. - Emit precise task-success asset event extra metadata: observed Delta version and table identity (documenting that intermediate commits may be coalesced, and authors requiring pinned reads must VERSION AS OF). - Support baseline_version, target_version, and allow_recreation handling. - Support deferrable execution via async triggerer polling. - Add provider documentation including reproducibility notes and Phase 1 scope callout. - Register sensor and trigger in provider.yaml and get_provider_info.py.
1 parent 11a6cba commit 9bb3d0e

7 files changed

Lines changed: 1179 additions & 1 deletion

File tree

‎providers/databricks/docs/operators/delta_table_sensor.rst‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
.. Licensed to the Apache Software Foundation (ASF) under one
2+
or more contributor license agreements. See the NOTICE file
3+
distributed with this work for additional information
4+
regarding copyright ownership. The ASF licenses this file
5+
to you under the Apache License, Version 2.0 (the
6+
"License"); you may not use this file except in compliance
7+
with the License. You may obtain a copy of the License at
8+
9+
.. http://www.apache.org/licenses/LICENSE-2.0
10+
11+
.. Unless required by applicable law or agreed to in writing,
12+
software distributed under the License is distributed on an
13+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
KIND, either express or implied. See the License for the
15+
specific language governing permissions and limitations
16+
under the License.
17+
18+
.. _howto/operator:DatabricksDeltaTableVersionSensor:
19+
20+
DatabricksDeltaTableVersionSensor (Phase 1)
21+
===========================================
22+
23+
Use the :class:`~airflow.providers.databricks.sensors.databricks_delta_table.DatabricksDeltaTableVersionSensor`
24+
to monitor a Delta table in Databricks and detect new commit versions. When a newer version is observed,
25+
the sensor succeeds, pushes provenance metadata to XCom, and optionally emits an Airflow Asset event.
26+
27+
The sensor can execute synchronously in standard worker slots or in deferrable mode using
28+
:class:`~airflow.providers.databricks.triggers.databricks_delta_table.DatabricksDeltaTableVersionTrigger`.
29+
30+
.. note::
31+
32+
**Scope (Phase 1 only):** This sensor provides task-level Delta table version observation inside a scheduled
33+
DAG task run. A continuous external asset producer with long-running coalescing across runs is planned as a
34+
follow-up Phase 2 feature.
35+
36+
37+
Using the Sensor
38+
----------------
39+
40+
The sensor requires either ``table_name`` or ``unity_table`` to identify the target Delta table:
41+
42+
* ``table_name`` - String representing the 1-, 2-, or 3-part Delta table name (``table``, ``schema.table``, or ``catalog.schema.table``).
43+
* ``unity_table`` - An instance of :class:`~airflow.providers.databricks.assets.databricks.UnityTableIdentity` specifying the host, catalog, schema, and table. When provided without explicit ``outlets``, the sensor automatically registers the corresponding Unity Catalog asset (``databricks://<host>/<catalog>/<schema>/<table>``) as an outlet.
44+
45+
One of the following Databricks SQL connection endpoints is required:
46+
47+
* ``sql_warehouse_name`` - Name of the Databricks SQL warehouse to execute queries against.
48+
* ``http_path`` - HTTP path of the Databricks SQL warehouse or interactive cluster.
49+
50+
Optional parameters:
51+
52+
* ``baseline_version`` - The baseline Delta version to compare against. If specified, the sensor succeeds when the table version is strictly greater than ``baseline_version``. If omitted (``None``) and ``target_version`` is also ``None``, the sensor automatically captures the table\x27s current version on the first poke and waits for subsequent commits (initial enrollment pattern).
53+
* ``target_version`` - An exact target Delta version to wait for.
54+
* ``allow_recreation`` - Boolean (default ``True``). When ``True``, if the table is dropped and recreated (causing the Delta version to reset to 0 or a number lower than ``baseline_version``), the sensor recognizes the recreation as a change and succeeds. If ``False``, a lower version raises an ``AirflowException``.
55+
* ``deferrable`` - Boolean (default ``conf.getboolean("operators", "default_deferrable", fallback=False)``). When ``True``, the sensor yields execution to the triggerer using ``DatabricksDeltaTableVersionTrigger``.
56+
* ``poke_interval`` - Time in seconds between successive status checks (default ``60`` seconds).
57+
* ``timeout`` - Maximum time in seconds to wait for a newer version before timing out (default 7 days).
58+
59+
60+
Asset Outlet Events and Delivery Semantics
61+
------------------------------------------
62+
63+
When the sensor declares outlets (or when ``unity_table`` automatically registers the Unity Catalog asset outlet),
64+
the task-success asset event carries precise semantics:
65+
66+
.. important::
67+
68+
The task-success asset event means **"a newer Delta version was observed"**, with the observed ``version``
69+
and ``table_identity`` included in the asset event ``extra`` dictionary.
70+
71+
* It is **not** a vague "table refreshed" event.
72+
* It is **not** a guarantee that every intermediate commit was delivered. If multiple commits occurred between
73+
polling checks, intermediate versions are coalesced and only the latest observed version is emitted.
74+
* **Pinned reads require explicit versioning:** Observing Delta table version *N* does **not** automatically pin
75+
downstream queries to version *N*. Downstream tasks or queries reading the table will query the head of the table
76+
unless explicitly pinned. Authors who need a reproducible, pinned read **must explicitly** specify
77+
``VERSION AS OF <observed_version>`` in their SQL queries (e.g. ``SELECT * FROM table VERSION AS OF {{ ti.xcom_pull(task_ids="sensor")["version"] }}``)
78+
or Delta Lake reader options (e.g. ``read.option("versionAsOf", version)``).
79+
80+
The asset event ``extra`` dictionary includes:
81+
82+
* ``version`` - The newly observed Delta table version (integer).
83+
* ``observed_version`` - Duplicate alias for ``version`` to make semantics unambiguous.
84+
* ``table_identity`` - A dictionary representing the table identity (e.g., ``{"host": ..., "catalog": ..., "schema": ..., "table": ...}``).
85+
* ``operation`` - The Delta operation that created the commit (e.g., ``WRITE``, ``MERGE``, ``CREATE OR REPLACE TABLE``).
86+
* ``timestamp`` - The commit timestamp from the Delta transaction log history (if available).
87+
88+
This metadata is also returned by the sensor and pushed to XCom under the key ``delta_table_version`` and ``return_value``.
89+
90+
91+
Examples
92+
--------
93+
94+
Waiting for table commits with baseline version:
95+
96+
.. code-block:: python
97+
98+
from airflow.providers.databricks.sensors.databricks_delta_table import (
99+
DatabricksDeltaTableVersionSensor,
100+
)
101+
102+
wait_for_updates = DatabricksDeltaTableVersionSensor(
103+
task_id="wait_for_table_updates",
104+
table_name="main.default.customers",
105+
sql_warehouse_name="my_warehouse",
106+
baseline_version=42,
107+
poke_interval=30,
108+
deferrable=True,
109+
)
110+
111+
Using Unity Catalog asset enrollment and downstream pinned read:
112+
113+
.. code-block:: python
114+
115+
from airflow.providers.common.compat.assets import Asset
116+
from airflow.providers.databricks.assets.databricks import UnityTableIdentity
117+
from airflow.providers.databricks.sensors.databricks_delta_table import (
118+
DatabricksDeltaTableVersionSensor,
119+
)
120+
121+
unity_table = UnityTableIdentity(
122+
host="my-workspace.cloud.databricks.com",
123+
catalog="analytics",
124+
schema="marketing",
125+
table="attribution_leads",
126+
)
127+
128+
wait_for_attribution = DatabricksDeltaTableVersionSensor(
129+
task_id="wait_for_attribution_leads",
130+
unity_table=unity_table,
131+
sql_warehouse_name="my_warehouse",
132+
deferrable=True,
133+
)
134+
135+
# Downstream DAG consuming the asset event and pinning reads to observed version:
136+
with DAG(
137+
dag_id="downstream_attribution_consumer",
138+
schedule=[unity_table.to_asset()],
139+
...,
140+
):
141+
...

‎providers/databricks/provider.yaml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,13 +186,15 @@ triggers:
186186
- integration-name: Databricks
187187
python-modules:
188188
- airflow.providers.databricks.triggers.databricks
189+
- airflow.providers.databricks.triggers.databricks_delta_table
189190

190191
sensors:
191192
- integration-name: Databricks
192193
python-modules:
193194
- airflow.providers.databricks.sensors.databricks
194195
- airflow.providers.databricks.sensors.databricks_sql
195196
- airflow.providers.databricks.sensors.databricks_partition
197+
- airflow.providers.databricks.sensors.databricks_delta_table
196198

197199
connection-types:
198200
- hook-class-name: airflow.providers.databricks.hooks.databricks.DatabricksHook

‎providers/databricks/src/airflow/providers/databricks/get_provider_info.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,10 @@ def get_provider_info():
121121
"triggers": [
122122
{
123123
"integration-name": "Databricks",
124-
"python-modules": ["airflow.providers.databricks.triggers.databricks"],
124+
"python-modules": [
125+
"airflow.providers.databricks.triggers.databricks",
126+
"airflow.providers.databricks.triggers.databricks_delta_table",
127+
],
125128
}
126129
],
127130
"sensors": [
@@ -131,6 +134,7 @@ def get_provider_info():
131134
"airflow.providers.databricks.sensors.databricks",
132135
"airflow.providers.databricks.sensors.databricks_sql",
133136
"airflow.providers.databricks.sensors.databricks_partition",
137+
"airflow.providers.databricks.sensors.databricks_delta_table",
134138
],
135139
}
136140
],

0 commit comments

Comments
 (0)