Page History

Status

Motivation

Proposed Changes

We propose extending the taskflow API to include asset-centric primitives to improve the interface for implementing workflows for asset-centric tools. In addition to attaching Asset annotations to a task-based workflow, as described in AIP-74, an asset can be defined directly in a file with a decorated Python function:

# The "at" argument is optional but strongly encouraged.
# The asset's name defaults to the function name if none is explicitly given.
@asset(at="s3://aws_conn_id@bucket/raw_bus_trips.parquet")
def raw_bus_trips():
    # Write bus trips data to asset...

The asset is at the same level as a DAG and defined by the function. Unlike @task and @dag, you do not need to call the decorated function in a file to instantiate it. The definition itself creates the asset. Calling the asset as a function results in a parse-time error.

The at argument specifies the asset’s location. It allows simply passing in a plain URI string or an ObjectStoragePath to be automatically coerced. We do not propose to actually handle writing to and reading from the URI—that is left as a future work at the moment—the user still needs to do that manually in the function.

All assets in an Airflow deployment share one single namespace. This is also the same namespace for DAGs to avoid confusion. Different assets can reference the same URI to support writing to the same data target from two functions—not a best but unfortunately common practice.

Naming an asset context results in a parse-time error. The name is reserved to reference the execution context provided by Airflow.

Reference an Asset

An asset can be referenced in another asset’s function by passing its name as a named function argument. 

@asset(at="snowflake://.../bus_trips")
def aggregated_bus_trips(raw_bus_trips):
    # Write aggregated bus trips to asset...

This does not control when the referencing asset is run. In the above example, aggregated_bus_trips can run regardless of the state of raw_bus_trips. If you want the referencing asset to depend on the referenced, the two must have a scheduling relationship. See Schedules below for details.

Multi-Assets

There are cases where the same function may generate multiple assets, such as when generating training and testing datasets for ML. This isn’t necessary for most situations, but is a valid use case (like an upstream asset needing to be split into two). To accommodate, we will allow users to pass in parameters for multiple assets. Some parameters will be unique per asset and some, like schedules or partitions (see below) will be shared.

@asset.multi(
    outlets=[
        Asset(name="bus_trip_1", aturi="s3://../bus_trip_1.parquet"),
        Asset(name="bus_trip_2", aturi="s3://../bus_trip_2.parquet"),
    ],
)
def split_trips(aggregated_bus_trips):
    ...

Schedules

The schedule of an asset denotes when the asset is written to. If the writing is done by a process managed by the pipeline, the schedule reflects when the pipeline kicks off the process to do the writing. Similar to today, you will also be able to set time-based schedules and/or schedule assets to be generated whenever upstream assets are generated. 

@asset(..., schedule="@yearly")
def asset1():  # Write happens when a new calendar year is entered.
    ...

@asset(..., schedule=asset1)
def asset2():  # Write happens whenever asset1’s write completes.
    ...

The key difference is, since we are decoupling partitioning from scheduling, the schedule parameter no longer controls the interval, i.e. not designed around the logical/execution date. It simply controls when the next round should happen.

It is expected that scheduling of non-asset workflows (DAGs) will also be changed in a similar way to match the behavior for assets. Existing operators must be reviewed to ensure they account for the new scheduling semantic, but we should provide a transition interface to assist rewrites and better allow providers to develop a implementation both compatible to 2 and 3. The timetable protocol will also require new methods to implement the new semantic, but both old and new should be able to both implemented on the same class, with each major version only calling methods its uses.

Partitions

At a high-level, partitions are “slices” of data assets that can be tracked and computed independently. They typically correspond to separate files or slices of tables, and are especially useful for modeling data that is appended or modified when running incremental load processes. When an asset is partitioned, each materialization of it only updates a part of the target, e.g. appends data newly generated during the time period covered by the partition. 

@asset(
    ...,
    schedule="@hourly",
    partition=PartitionByInternal("@hourly"),  # Each materialization covers an hour.
)
def hourly_data():
    ...

The partition concept and interface is covered by AIP-76.

Scope Definition

This AIP is considered complete when the @asset decorator is implemented and can fulfill the above features. Scheduling and partitioning support not related to assets, such as partition support in DAGs, is not a part of this AIP.