@asset syntax in Apache Airflow®

1
from airflow.sdk import asset
2
3
4
@asset(schedule="@daily")
5
def extracted_data():
6
    return {"a": 1, "b": 2}
7
8
9
@asset(schedule=extracted_data)
10
def transformed_data(context):
11
12
    data = context["ti"].xcom_pull(
13
        dag_id="extracted_data",
14
        task_ids="extracted_data",
15
        key="return_value",
16
        include_prior_dates=True,
17
    )
18
    return {k: v * 2 for k, v in data.items()}
19
20
21
@asset(schedule=transformed_data)
22
def loaded_data(context):
23
24
    data = context["task_instance"].xcom_pull(
25
        dag_id="transformed_data",
26
        task_ids="transformed_data",
27
        key="return_value",
28
        include_prior_dates=True,
29
    )
30
    summed_data = sum(data.values())
31
    print(f"Summed data: {summed_data}")

1
from airflow.sdk import Asset, dag, task
2
3
4
@dag(schedule="@daily")
5
def extract_dag():
6
7
    @task(outlets=[Asset("extracted_data")])
8
    def extract_task():
9
        return {"a": 1, "b": 2}
10
11
    extract_task()
12
13
14
extract_dag()
15
16
17
@dag(schedule=[Asset("extracted_data")])
18
def transform_dag():
19
20
    @task(outlets=[Asset("transformed_data")])
21
    def transform_task(**context):
22
        data = context["ti"].xcom_pull(
23
            dag_id="extract_dag",
24
            task_ids="extract_task",
25
            key="return_value",
26
            include_prior_dates=True,
27
        )
28
        return {k: v * 2 for k, v in data.items()}
29
30
    transform_task()
31
32
33
transform_dag()
34
35
36
@dag(schedule=[Asset("transformed_data")])
37
def load_dag():
38
39
    @task
40
    def load_task(**context):
41
        data = context["ti"].xcom_pull(
42
            dag_id="transform_dag",
43
            task_ids="transform_task",
44
            key="return_value",
45
            include_prior_dates=True,
46
        )
47
        summed_data = sum(data.values())
48
        print(f"Summed data: {summed_data}")
49
50
    load_task()
51
52
53
load_dag()