anyone using datahub lineage with Airflow in Google Composer? I have installed

acryl-datahub[airflow]==0.8.6.4

as a package but it is not showing connection type in airflow UI

@breezy-glass-7892 has used datahub lineage with Google Composer if I'm not mistaken

Yeah, that’s right.

@breezy-glass-7892 you used cli to create the connection? It is not showing in the UI Also, google composer shows this in configuration override. So does lineage work?

The configuration properties are stored in a file called airflow.cfg the first time you run Apache Airflow. You can choose to change these properties and override the default values. Sections include: core, cli, api, operators, hive, webserver, email, smtp, celery, celery_broker_transport_options, dask, scheduler, ldap, mesos, kerberos, github_enterprise, admin.

If yes, how did you add indentation? https://datahubproject.io/docs/metadata-ingestion#using-datahubs-airflow-lineage-backend-recommended. This is a simple textbox. Not really sure how to add indentation in here

Are you talking about establishing a connection with the Daatahub service?

yes for lineage backend

You have to first create an ingress service on GCP

I am not talking about the access to the UI of datahub. That I already have.

Under your airflow UI, just go to Admin -> Connections and add a new connection

Are you running airflow 1 or 2? The datahub connection types are not showing up for me

Yeah you can leave that blank

How did you add this block in config? It is a textbox https://datahubspace.slack.com/archives/C029A3M079U/p1627969008023400?thread_ts=1627630569.003400&cid=C029A3M079U

I have not updated the

airflow.cfg

file

Lineage is working for you even when you have not specified the lineage backend?

Yea, I think this link is leveraging the Airflow lineage info

Ok. Thanks @breezy-glass-7892

Hi @square-activity-64562: I think for google composer maybe go with the a la carte operator option.

I was actually talking about having a CLI. This allows us to create a kubernetes Pod in a separate cluster. Using code from https://github.com/linkedin/datahub/blob/master/metadata-ingestion/src/datahub_provider/_lineage_core.py I came up with this as a draft.

import click
import datahub.emitter.mce_builder as builder
import datahub.metadata.schema_classes as models
from datahub.emitter.rest_emitter import DatahubRestEmitter
from typing import List

PLATFORMS = [
    'athena', 
    'postgres'
]

def _validate_platform(platform):
    if platform not in PLATFORMS:
        raise RuntimeError(f"Unknown platform ${platform} passed in urn")


def _get_dataset_urn(string: str):
    platform, dataset_id = string.split(":")
    
    _validate_platform(platform)

    return builder.make_dataset_urn(platform, dataset_id)


def send_lineage(gms_url, base_url, dag_id, task_id, upstream_urns: List[str], downstream_urns: str):

    input_datasets=[_get_dataset_urn(string) for string in upstream_urns]
    output_datasets=[_get_dataset_urn(string) for string in downstream_urns]

    flow_urn = builder.make_data_flow_urn("airflow", dag_id)
    job_urn = builder.make_data_job_urn_with_flow(flow_urn, task_id)

    flow_url = f"{base_url}/airflow/tree?dag_id={dag_id}"
    job_url = f"{base_url}/taskinstance/?flt0_dag_id_equals={dag_id}&flt3_task_id_equals={task_id}"

    flow_mce = models.MetadataChangeEventClass(
        proposedSnapshot=models.DataFlowSnapshotClass(
            urn=flow_urn,
            aspects=[
                models.DataFlowInfoClass(
                    name=dag_id,
                    externalUrl=flow_url,
                )
            ],
        )
    )

    job_mce = models.MetadataChangeEventClass(
        proposedSnapshot=models.DataJobSnapshotClass(
            urn=job_urn,
            aspects=[
                models.DataJobInfoClass(
                    name=task_id,
                    type=models.AzkabanJobTypeClass.COMMAND,
                    externalUrl=job_url,
                ),
                models.DataJobInputOutputClass(
                    inputDatasets=input_datasets,
                    outputDatasets=output_datasets,
                    inputDatajobs=[],
                )
            ],
        )
    )
    force_entity_materialization = [
        models.MetadataChangeEventClass(
            proposedSnapshot=models.DatasetSnapshotClass(
                urn=iolet,
                aspects=[
                    models.StatusClass(removed=False),
                ],
            )
        )
        for iolet in input_datasets + output_datasets
    ]
    mces = [
        flow_mce,
        job_mce,
        *force_entity_materialization
    ]

    emitter = DatahubRestEmitter(gms_url)

    for mce in mces:
        emitter.emit_mce(mce)


@click.command()
@click.option('--gms-url', required=True)
@click.option('--airflow-url', required=True)
@click.option('--dag-id', required=True)
@click.option('--task-id', required=True)
@click.option('--upstream-urn', required=True, multiple=True)
@click.option('--downstream-urn', required=True, multiple=True)
def send_job_lineage(gms_url, airflow_url, dag_id, task_id, upstream_urn, downstream_urn):
    send_lineage(
        gms_url,
        airflow_url,
        dag_id,
        task_id,
        list(upstream_urn),
        list(downstream_urn)
    )

send_job_lineage()