この記事でやること

この記事ではMicrosoft PurviewのLineageをREST APIでJSON形式で取得しGraphvizを使って可視化します。
REST APIの実行はPythonを使っています。

環境など

Microsoft Purview REST API
PurviewのGet Lineage By Unique AttributeでLineageがJSON形式で返されます。

GET {Endpoint}/catalog/api/atlas/v2/lineage/uniqueAttribute/type/{typeName}?direction={direction}

graphviz
JSONで返されたLineageをGraphvizのPythonインターフェイスを使ってSVG表示します。

サンプルLineage

今回こんな感じのLineageを出力していきます。
　S3->ADF(Copy)->ADLS->ADF(DataFlow)->SQL Databse

このLineageをREST APIで取得するとこんな構造の出力が得られます。

頭が痛くなりますね。。

Lineageの取得と表示

ここからは実際にLineageの取得と表示を行っていきます。

アクセストークンの取得

import json
import requests
from pprint import pprint

def azure_ad_auth(tenant_id: str, client_id: str, client_secret: str, resource_url: str):
    url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
    payload= f"grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}&resource={resource_url}"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    access_token = json.loads(response.text)['access_token']
    
    return access_token

tenant_id = "テナントID"
client_id = "アプリケーションID"
client_secret = "シークレット"
resource_url = "https://purview.azure.net"
data_catalog_name = "Purviewアカウント名"

azure_ad_access_token = azure_ad_auth(tenant_id, client_id, client_secret, resource_url)

Lineageの取得

今回はLineageの終端であるSQL DatabaseのCustomerInfoテーブルを起点にしてLineageを取得します。CustomerInfoテーブルのqualifiedNameとtypeNameを指定して取得します。
GUIDを指定してLineageを取得することもできます。

def get_lineage_by_atter(typeName: str, qualifiedName: str, direction: str):
    url = f"https://{data_catalog_name}.purview.azure.com/catalog/api/atlas/v2/lineage/uniqueAttribute/type/{typeName}?attr:qualifiedName={qualifiedName}"
    url += f"&direction={direction}"

    headers = {
        "Authorization": "Bearer " + azure_ad_access_token,
        "Content-Type": "application/json"
    }
    response = json.loads(requests.get(url, headers=headers).content)

    return response

direction = "BOTH" # Lineageを取得する方向
type_name = "azure_sql_table" # SQLDBのテーブルなのでazure_sql_tableを指定
qualified_name = "" # 下記参照

response = get_lineage_by_atter(type_name, qualified_name, direction)

qualifiedNameは↓から取得します。

Lineageの表示（その1）

今回、まずはguidEntityMapに着目します。

guidEntityMapにはLineageの青い枠で表現されている部分の情報が入っています。（線に当たる部分はありません。）

まずは、このノードの部分をGraphvizで表示します。

from graphviz import Digraph
from IPython.display import SVG

lineage01 = Digraph(format="svg")
lineage01.attr('graph', rankdir="LR", compound="true")
lineage01.attr("node", fontname="Segoe UI")

for guid in response["guidEntityMap"]:
    lineage01.node(
        guid,
        shape="box",
        label=f'''
{guid}
{response["guidEntityMap"][guid]["attributes"]["name"]}
{response["guidEntityMap"][guid]["typeName"]}
        '''
    )

lineage01.render("lineage/01.gv")
SVG("lineage/01.gv.svg")

GUID、エンティティ名、タイプ名などが表示できました。

おまけ: GUIDを指定してLineageを取得する

def get_lineage_by_guid(guid: str, direction: str):
    url = f"https://{data_catalog_name}.purview.azure.com/catalog/api/atlas/v2/lineage/{guid}?direction={direction}"

    headers = {
        "Authorization": "Bearer " + azure_ad_access_token,
        "Content-Type": "application/json"
    }
    response = json.loads(requests.get(url, headers=headers).content)

    return response

guid = "b8ff9bf2-5ee6-4849-b3ad-93f6f6f60000"
direction = "BOTH"
response = get_lineage_by_guid(guid, direction)

GUIDはURLから取得します。

次回はノードを繋いでいきます。