[WIP] AIP-75 New Asset-Centric Syntax

Status

Motivation

Rationale

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 the use case of 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.

Asset References

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", at="s3://../bus_trip_1.parquet"),
        Asset(name="bus_trip_2", at="s3://../bus_trip_2.parquet"),
    ],
)
def split_trips(aggregated_bus_trips):
    ...