> ## Documentation Index
> Fetch the complete documentation index at: https://docs.getcollate.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Airflow Configuring Lineage

> Learn how to configure DAG lineage in Airflow using inlets and outlets. Supports tables, dashboards, pipelines, and more with or without ingestion packages.

# Configuring DAG Lineage

Regardless of the Airflow ingestion process you follow ([Workflow](/ai-2-0/connectors/pipeline/airflow),
[Lineage Backend](/ai-2-0/connectors/pipeline/airflow/lineage-backend) or [Lineage Operator](/ai-2-0/connectors/pipeline/airflow/lineage-operator)),
Collate will try to extract the lineage information based on the tasks `inlets` and `outlets`.

What's important to consider here is that when Collate ingests Airflow lineage, it is actually building a graph:

```
Table A (node) -> DAG (edge) -> Table B (node)
```

Where tables are nodes and DAGs (Pipelines) are considered edges. This means that the correct way of setting these
parameters is by making sure that both `inlets` and `outlets` are informed, so that Collate has the nodes to build
the relationship.

## Configuring Lineage

<Tip>
  **Tip**: Collate supports lineage for the following entities: `Table`, `Container`, `Dashboard`, `DashboardDataModel`, `Pipeline`, `Topic`, `SearchIndex`, `REST API`, and `MlModel`.

  Moreover, note that this example requires the `openmetadata-ingestion` package to be installed. If you plan to
  ingest the Airflow metadata (and lineage) externally and don't want to install it, see the next section.
</Tip>

Take a look at the following example:

```python theme={null}
from datetime import timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

from metadata.generated.schema.entity.data.container import Container
from metadata.generated.schema.entity.data.table import Table
from metadata.ingestion.source.pipeline.airflow.lineage_parser import OMEntity


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=1),
}


with DAG(
    "test-lineage",
    default_args=default_args,
    description="An example DAG which runs a lineage test",
    start_date=days_ago(1),
    is_paused_upon_creation=False,
    catchup=False,
) as dag:


    t0 = DummyOperator(
        task_id='task0',
        inlets=[
            OMEntity(entity=Container, fqn="Container A", key="group_A"),
            OMEntity(entity=Table, fqn="Table X", key="group_B"),
        ]
    )

    t1 = DummyOperator(
        task_id='task10',
        outlets=[
            OMEntity(entity=Table, fqn="Table B", key="group_A"),
            OMEntity(entity=Table, fqn="Table Y", key="group_B"),
        ]
    )

    t0 >> t1
```

This example passes inlets and outlets as a list of the `OMEntity` class, which lets you specify:

1. The type of the asset being used, such as Table or Container, following the Collate SDK.
2. The FQN of the asset, which is the unique name of each asset in Collate, for example, `serviceName.databaseName.schemaName.tableName`.
3. The key to group the lineage if needed.

This `OMEntity` class is defined following the example of Airflow's internal lineage
[models](https://github.com/apache/airflow/blob/2.9.0/airflow/lineage/entities.py).

## Keys

Specify the lineage dependencies among different groups of tables. In the example above, the lineage is not built
from all inlets to all outlets, but rather grouped by key (`group_A` and `group_B`).
This means that after this lineage is processed, the relationship will be:

```
Container A (node) -> DAG (edge) -> Table B (node)
```

and

```
Table X (node) -> DAG (edge) -> Table Y (node)
```

It does not matter in which task of the DAG these inlet/outlet information is specified. During the ingestion process,
Collate groups all these details at the DAG level.

## Configuring Lineage Without the `openmetadata-ingestion` Package

Apply the same example as above, but describe the lineage in dictionaries instead, to avoid requiring
the `openmetadata-ingestion` package to be installed in the environment.

```python theme={null}
from datetime import timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=1),
}


with DAG(
    "test-lineage",
    default_args=default_args,
    description="An example DAG which runs a lineage test",
    start_date=days_ago(1),
    is_paused_upon_creation=False,
    catchup=False,
) as dag:


    t0 = DummyOperator(
        task_id='task0',
        inlets=[
            {"entity": "container", "fqn": "Container A", "key": "group_A"},
            {"entity": "table", "fqn": "Table X", "key": "group_B"},
        ]
    )

    t1 = DummyOperator(
        task_id='task10',
        outlets=[
            {"entity": "table", "fqn": "Table B", "key": "group_A"},
            {"entity": "table", "fqn": "Table Y", "key": "group_B"},
        ]
    )

    t0 >> t1
```

This example passes inlets and outlets as a list of dictionaries, which lets you specify:

1. The type of the asset being used, following the list below.
2. The FQN of the asset, which is the unique name of each asset in Collate, for example, `serviceName.databaseName.schemaName.tableName`.
3. The key to group the lineage if needed.

The `entity` key needs to be informed as follows for each of the entity types:

* Table: `table`
* Container: `container`
* Dashboard: `dashboard`
* Dashboard Data Model: `dashboardDataModel`
* Pipeline: `pipeline`
* Topic: `topic`
* SearchIndex: `searchIndex`
* MlModel: `mlmodel`

<Tip>
  **Tip**: When configuring Airflow lineage without the `openmetadata-ingestion` package, **only table entities** are supported using the simplified format:

  ```python theme={null}
  inlets = [{"tables": ["service.db.schema.table"]}]
  ```

  * Dashboards, topics, and other non-table entities are not supported in this format.
  * To use those, you must use the full `OMEntity` format, which requires the `openmetadata-ingestion` package.
</Tip>

## Configuring Lineage Between Tables

<Tip>
  **Tip**: Note that this method only allows lineage between tables.

  This method will be deprecated in Collate 1.4.
</Tip>

Take a look at the following example:

```python theme={null}
from datetime import timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=1),
}


with DAG(
    "test-multiple-inlet-keys",
    default_args=default_args,
        description="An example DAG which runs a lineage test",
    start_date=days_ago(1),
    is_paused_upon_creation=False,
    catchup=False,
) as dag:


    t0 = DummyOperator(
        task_id='task0',
        inlets={
            "group_A": ["Table A"],
            "group_B": ["Table X"]
        }
    )

    t1 = DummyOperator(
        task_id='task10',
        outlets={
            "group_A": ["Table B"],
            "group_B": ["Table Y"]
        }
    )

    t0 >> t1
```

<Tip>
  **Tip**: Make sure to add the table Fully Qualified Name (FQN), which is the unique name of the table in Collate.

  This name is composed as `serviceName.databaseName.schemaName.tableName`.
</Tip>