Configuring DAG Lineage

Regardless of the Airflow ingestion process you follow (Workflow, Lineage Backend or 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: 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.

Take a look at the following example:

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:

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

This OMEntity class is defined following the example of Airflow’s internal lineage models.

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.

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:

The type of the asset being used, following the list below.
The FQN of the asset, which is the unique name of each asset in Collate, for example, serviceName.databaseName.schemaName.tableName.
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: When configuring Airflow lineage without the openmetadata-ingestion package, only table entities are supported using the simplified format:

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.

Configuring Lineage Between Tables

Tip: Note that this method only allows lineage between tables.This method will be deprecated in Collate 1.4.

Take a look at the following example:

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: 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.

Connectors

Connectors

Airflow Configuring Lineage

Configuring DAG Lineage

Configuring Lineage

Keys

Configuring Lineage Without the `openmetadata-ingestion` Package

Configuring Lineage Between Tables

​Configuring DAG Lineage

​Configuring Lineage

​Keys

​Configuring Lineage Without the openmetadata-ingestion Package

​Configuring Lineage Between Tables

Configuring DAG Lineage

Configuring Lineage

Keys

Configuring Lineage Without the `openmetadata-ingestion` Package

Configuring Lineage Between Tables