Data Lineage Producer Python Library

Python producer library for Data Lineage.

The library provides synchronous and asynchronous clients for interacting with the Data Lineage API.

Installation

Requires Python 3.9 or later.

To install using pip:

pip install google-cloud-datacatalog-lineage-producer-client

Before you begin

1. Enable the Data Lineage API in your Google Cloud project.

2. Make sure your principal has a role that allows reporting lineage, such as roles/datalineage.producer or roles/datalineage.editor.

3. Set up Application Default Credentials:

   gcloud auth application-default login
   

Composing an OpenLineage message

open_lineage_message = {
    "eventType": "COMPLETE",
    "eventTime": "2023-04-04T13:21:16.098Z",
    "producer": "producer-1",
    "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/$defs/RunEvent",
    "job": {
        "namespace": "namespace-1",
        "name": "job-1",
    },
    "run": {
        "runId": "1234-5678",
    },
    "inputs": [
        {
            "namespace": "bigquery",
            "name": "input_project.input_dataset.input_table",
        }
    ],
    "outputs": [
        {
            "namespace": "bigquery",
            "name": "output_project.output_dataset.output_table",
        }
    ],
}

or, to convert it from an existing JSON string:

import json

open_lineage_message = json.loads(json_string)

Performing calls

To perform a call, you need to:

1. Create a client
2. Create a request object
3. Call a corresponding method

parent is the project and location that the lineage is reported to, in the format projects/PROJECT_ID/locations/LOCATION.

from google.cloud.datacatalog.lineage.producer_client.v1 import SyncLineageClient
from google.cloud.datacatalog_lineage_v1 import ProcessOpenLineageRunEventRequest

client = SyncLineageClient()
parent = "projects/PROJECT_ID/locations/LOCATION"
request = ProcessOpenLineageRunEventRequest(
    parent=parent,
    open_lineage=open_lineage_message,
)
response = client.process_open_lineage_run_event(request)
print(response)

For asynchronous calls:

import asyncio

from google.cloud.datacatalog.lineage.producer_client.v1 import AsyncLineageClient
from google.cloud.datacatalog_lineage_v1 import ProcessOpenLineageRunEventRequest

async def main():
    client = AsyncLineageClient()
    parent = "projects/PROJECT_ID/locations/LOCATION"
    request = ProcessOpenLineageRunEventRequest(
        parent=parent,
        open_lineage=open_lineage_message,
    )
    response = await client.process_open_lineage_run_event(request)
    print(response)

asyncio.run(main())

The response contains the names of the created process, run and lineage events.