この記事でやること
この記事では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から取得します。
次回はノードを繋いでいきます。