|
Incremental processing is a very common pattern in Airflow use cases. In some pipelines, this is strictly necessary. Although external event driven scheduling is supported today in Airflow through AIP-82, incremental polling is not natively supported. In some scenarios, users are forced to use XCom or Variables handling the watermark for incremental processing, or inventing and implementing their own solution. In the case of event-driven triggering, it’s even more challenging for users to build their custom incremental processing implantation, due to the lack of XCom and Variable support.
There have been several attempts to store state within a child of the BaseEventTrigger to manage incremental processes, while none of these have proven to be effective or robust. This has seriously hampered the ability for the community to build logic to monitor Assets such as object stores, SQL databases, and other non-queue/stream-based Assets.
For Triggers built for Asset “watching”, it is helpful, if not essential to persist some state.
At a high level, we propose to make the following changes:
asset_watermarks model for persisting Asset/Trigger state metadata. This model would allow for Airflow users to store and retrieve state as specified in EventTrigger code. A record (watermark) stored asset_watermarks contains an asset_name and namespace , as well as a pair of key and value. The key must be unique in a namespace.This solves one of the most glaring problems with building Triggers compatible with "Asset-watching", or event-driven scheduling; the inability to persist state. Without a model like asset_watermarks , it's nearly impossible to do things like monitor an S3 bucket for new files landing or handle the addition/removal of a new row to a SQL database. Despite being one of the most touted features of Airflow 3.0, the community has been very slow to develop and distribute Event Triggers to be used for "Asset-watching". This AIP aims to address this problem in an elegant and "Airflow-onic" way, using a "batteries included" model, along with a pluggable backend.
Building the asset_watermarks model would require minimal lift. Like with the Variable model, this would be “unrelated” to other models, making the creation a bit easier. This model would include four fields; an Asset name, namespace, a key, and a value. The namespace would allow for user-specified identification between two Assets with the same name. Below is the pseudo-code for the asset_watermarks model.
class AssetWatermark:
"""
Watermark class allows pluggable db backend.
"""
def __init__(self, asset_name: str, namespace: str, key: str):
self.asset_name = asset_name
self.namespace = namespace
self.key = key
def set_value(self, value):
db_backend.set_value(asset_name=self.asset_name, namespace=self.namespace, watermark_key=self.key, watermark_value=value)
def get_value(self):
return db_backend.get_value(asset_name=self.asset_name, namespace=self.namespace, watermark_key=self.key) |
For users who plan on using asset_watermarks extensively, a pluggable backend might provide them more control over the actual storage resources used to persist state. This backend would closely mimic the XComObjectStorageBackend, which includes an out-of-the-box model with the ability to configure a “pluggable” backend.
It would require some sort of get and set logic to be implemented in a class that inherits from BaseEventTrigger in order to store and retrieve watermark value. This would most likely be in the form of a method for each as part of the BaseEventTrigger class, as shown in below pseudo-code.
class BaseEventTrigger(BaseTrigger):
@property
def asset_watermark(self):
return AssetWatermark(asset_name=self.asset_name, namespace=self.watermark_namespace, key=self.watermark_key)
@abstractmethod
def get_high_watermark(self, *args, **kwargs):
# put the logic to calculate the watermark here
pass
def persist_watermark(self, value):
self.asset_watermark.set_value(value=value)
@property
def low_watermark(self):
return self.asset_watermark.get_value() |
One of the most common use-cases for event-driven scheduling will most likely be “watching” an object store for changes. Ideally, each time that an AssetWatcher’s Trigger runs, it will not re-scan the entire bucket. The workflow would look something like this:
Another common use-case for event-driven scheduling is “watching” tables in relational databases; we’ll use Postgres as an example. Outside of traditional CDC, it’s common for data teams to use an updated column in a Postgres database to upsert data. Using the asset_watermark approach, they can use the following workflow. This should feel quite similar to the workflow for “watching” an S3 Bucket; that’s intentional, as we’re trying to implement a repeatable pattern.
Below is an example of a Trigger that executes a SQL query to incrementally read data, with a watermark.
class SQLIncrementalQueryTrigger(BaseEventTrigger):
...
def get_high_watermark(self, *args, **kwargs):
return pendulum.now().to_date_string()
async def run(self):
while True:
results = conn.get_records("SELECT * FROM table WHERE updated_at > '%s'", self.low_watermark)
self.persist_watermark(key="table_last_updated_at", value=self.get_high_watermark())
if results:
yield TriggerEvent({"status": "success", "results": results})
break
else:
await asyncio.sleep(self.poke_interval) |
Allow for an Airflow-supported methodology for persisting watermarks and unblocking the development of Event Triggers for Airflow users interested in "Asset-watching".
There is one theorized downside; Airflow users leveraging the asset_watermarks model as a sort of key-value store outside of an Event Trigger. However, this is not something that is overly concerning. It would be much more likely for Airflow users to abuse the Variable model instead. Along with this are the general downsides to added complexity, but those are minimal with this proposal.
DAG Authors: these users will now have access to a tool that makes authoring Triggers, Sensors, and Tasks used to orchestrate incremental processes more accessible and unlocks the ability to further build out event-driven logic.
Deployment Managers: introducing a pattern with a pluggable backend may provide deployment managers an additional component of their Airflow “stack” to stand up and manage. This will also require a DB migration, which is outlined below.
Current users who don't use incremental event triggers are unaffected by this change. However, this new feature requires a DB migration, since a new model is being added. The only changes needed in this case would be the addition of a new table.
There are not breaking changes included as part of this AIP. Added the assets_watermark model will require a DB migration, but forces no breaking changes upon users.
This AIP will be considered “done” when the PR creates the model and the pluggable backend is merged and the solution is well-docmented.