Note: Lineage support is very experimental and subject to change.

    Airflow can help track origins of data, what happens to it and where it moves over time. This can aid having audit trails and data governance, but also debugging of data flows.

    Airflow tracks data by means of inlets(入口) and outlets of the tasks. Let’s work from an example and see how it works.

    1. from airflow.operators.bash import BashOperator
    2. from airflow.operators.dummy import DummyOperator
    3. from airflow.lineage import AUTO
    4. from airflow.lineage.entities import File
    5. from airflow.models import DAG
    6. from airflow.utils.dates import days_ago
    7. from datetime import timedelta
    9. FILE_CATEGORIES = ["CAT1", "CAT2", "CAT3"]
    11. args = {
    12.     'owner': 'airflow',
    13.     'start_date': days_ago(2)
    14. }
    16. dag = DAG(
    17.     dag_id='example_lineage', 
    18.     default_args=args,
    19.     schedule_interval='0 0 * * *',
    20.     dagrun_timeout=timedelta(minutes=60))
    22. f_final = File(url="/tmp/final")
    23. run_this_last = DummyOperator(
    24.     task_id='run_this_last', 
    25.     dag=dag,
    26.     inlets=AUTO,
    27.     outlets=f_final)
    29. f_in = File(url="/tmp/whole_directory/")
    30. outlets = []
    31. for file in FILE_CATEGORIES:
    32.     f_out = File(url="/tmp/{}/{{{{ execution_date }}}}".format(file))
    33.     outlets.append(f_out)
    35. run_this = BashOperator(
    36.     task_id='run_me_first', 
    37.     bash_command='echo 1', 
    38.     dag=dag,
    39.     inlets=f_in,
    40.     outlets=outlets
    41. )
    42. run_this.set_downstream(run_this_last)


    Inlets can be a (list of) upstream task ids or statically defined as an attr annotated object as is, for example, the File object. Outlets can only be attr annotated object. Both are rendered at run time. However the outlets of a task in case they are inlets to another task will not be re-rendered for the downstream task.