この記事でやること

この記事ではMicrosoft PurviewのLineageをREST APIでJSON形式で取得しGraphvizを使って可視化します。
REST APIの実行はPythonを使っています。
※今回のカラム取得方法はイマイチな感じがしてます。より効率的なカラム取得方法を見つけたらアップデートします。（逆に効率的な方法をご存じの方がいたら教えてください！）

Lineageの取得と表示

その1で取得したLineageをそのまま使っていきます。
今回はカラムの情報を追加します。

カラムマッピングの取得

その1で各エンティティの情報をguidEntityMapから取得していますが、その中のresponse["guidEntityMap"][guid]["attributes"]["columnMapping"]にカラム毎のマッピング情報が格納されています。

DatasetMappingにSourceとSinkのqualifiedNameが格納されていて、columnMappingにSourceとSinkのカラム名が配列で格納されている感じです。

Lineageとカラム情報をまとめる

Lineageを表示するための情報が多くなってきたのでGraphvizで使用する情報をentitiesにまとめています。

import pandas as pd

entities = {} # Graphvizで使用するデータの格納用
column_mappings = {}
pd.options.display.max_colwidth = 150
df_colum_mapping = pd.DataFrame(index=[], columns=["Source", "Sink"])

for guid in response["guidEntityMap"]:
    entities[guid] = {
        "guid": guid,
        "qualifiedName": response["guidEntityMap"][guid]["attributes"]["qualifiedName"],
        "name": response["guidEntityMap"][guid]["attributes"]["name"],
        "typeName": response["guidEntityMap"][guid]["typeName"],
        "classificationNames": response["guidEntityMap"][guid]["classificationNames"]
    }

    # columnMappingを持っているとき
    if response["guidEntityMap"][guid]["attributes"].get("columnMapping") is not None:
        # textで持っているのでJSONに変換
        json_column_mapping = json.loads(response["guidEntityMap"][guid]["attributes"]["columnMapping"])

        # SourceとSinkのqualifiedNameとカラム情報をDataFrameにまとめる
        for column_mapping in json_column_mapping:
            df_tmp = pd.json_normalize(column_mapping["ColumnMapping"])
            df_tmp = df_tmp.assign(SourceDataset=column_mapping["DatasetMapping"]["Source"])
            df_tmp = df_tmp.assign(SinkDataset=column_mapping["DatasetMapping"]["Sink"])
            df_colum_mapping = pd.concat([df_colum_mapping, df_tmp], ignore_index=True)

            # qualifiedNameを持っていないエンティティがあるため、GUIDから取得
            for relation in response["relations"]:
                if relation["toEntityId"] == guid:
                    if column_mapping["DatasetMapping"]["Source"] == "*":
                        df_colum_mapping.replace("*", response["guidEntityMap"][relation["fromEntityId"]]["attributes"]["qualifiedName"], inplace=True)

# SourceとSinkをまとめる
df_colum_mapping = pd.concat(
    [
        df_colum_mapping[["SourceDataset", "Source"]].rename(columns={"SourceDataset": "Dataset", "Source": "Column"}),
        df_colum_mapping[["SinkDataset", "Sink"]].rename(columns={"SinkDataset": "Dataset", "Sink": "Column"})
    ], 
    ignore_index=True)

# 重複排除
df_colum_mapping.drop_duplicates(inplace=True)
# display(df_colum_mapping)

# エンティティ毎にカラム情報を追加
for guid in entities:
    df_columns = df_colum_mapping.query(f'Dataset == "{entities[guid]["qualifiedName"]}"')
    df_columns = df_columns["Column"]
    entities[guid]["columns"] = df_columns.values.tolist()

Lineageの表示（その3）

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

for guid in entities:
    lineage03.node(
        guid,
        shape="box",
        label=f'''
{guid}
{entities[guid]["name"]}
{entities[guid]["typeName"]}
{entities[guid]["columns"] if len(entities[guid]["columns"]) > 0 else ""}
        '''
    )

for relation in response["relations"]:
    lineage03.edge(relation["fromEntityId"], relation["toEntityId"])

lineage03.render("lineage/03.gv")
SVG("lineage/03.gv.svg")

こんな感じでカラム情報もLineageに追加できました。

おまけ

ここからはGraphvizの話になります。
必要な情報は表示できましたが、モノクロで寂しいので色とかをつけて見やすくしていきます。
Graphvizではcssなども使えるようですが、今回使用しているノードのlabel情報には適用できないようなのでタグ内の要素にベタ打ちになっています。。

参考: GraphvizのHTML-Like Labels

Datasetの場合は四角に、Processの場合は角を丸くしたいのでentitiesに情報を加えておきます。（コードに書いてはいませんが、ここでアイコンのセットもしています。）

for guid in entities:
    # ProcessかDatasetかを設定
    if entities[guid]["typeName"].startswith("adf_"):
        entities[guid]["entityType"] = "process"
    else:
        entities[guid]["entityType"] = "dataset"

オリジナルのLineageには分類は表示されませんが情報としては持っているので、表示するようにしてみます。

lineage04 = Digraph(format="svg")
lineage04.attr("graph", rankdir="LR", compound="true")
# 背景色
lineage04.attr("graph", bgcolor="whitesmoke")
# ノードの枠線とフォントの設定
lineage04.attr("node", fontname="Segoe UI", fontcolor="#333333", color="dodgerblue")
# 矢印の色
lineage04.attr("edge", color="dimgray")

icon_dir = "icon/"
for guid in entities:
    entityInfo = ""

    # ADFのCopyアクティビティ、ADF名とPipeline名をセット
    if entities[guid]["typeName"] == "adf_copy_operation":
        entityInfo = f'''
    <tr>
        <td align="left"><font point-size="8">Data Factory:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[8]}</font></td></tr>
    <tr>
        <td align="left"><font point-size="8">Pipeline:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[10]}</font></td>
    </tr>
        '''
    # ADFのDataflow
    if entities[guid]["typeName"] == "adf_dataflow_operation":
        entityInfo = f'''
    <tr>
        <td align="left"><font point-size="8">Data Factory:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[-5]}</font></td></tr>
    <tr>
        <td align="left"><font point-size="8">Pipeline:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[-3]}</font></td>
    </tr>
        '''

    # SQL Databseのテーブルの場合は、Server名やDB名をセット
    if entities[guid]["typeName"] == "azure_sql_table":
        entityInfo = f'''
    <tr>
        <td align="left"><font point-size="8">Server:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[2]}</font></td>
    </tr>
    <tr>
        <td align="left"><font point-size="8">Database:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[3]}</font></td>
    </tr>
    <tr>
        <td align="left"><font point-size="8">Schema:</font></td>
        <td align="left"><font point-size="8">{entities[guid]["qualifiedName"].split("/")[4]}</font></td>
    </tr>
        '''

    # カラムをセット
    columns = ""
    for column in entities[guid]["columns"]:
        columns += f'''
    <tr><td align="left" colspan="3">    {column}</td></tr>
        '''

    # 分類をセット
    classifications = ""
    for classification in entities[guid]["classificationNames"]:
        classifications += f'''
    <font point-size="6" color="deeppink"><b>{classification.split(".")[-1]}</b></font>
        '''

    lineage04.node(
        guid,
        shape="plaintext",
        label=f'''<
<table border="1" cellspacing="1" cellborder="0" align="left"
    style="{"radial" if entities[guid]["entityType"] == "dataset" else "rounded"}"
    bgcolor="{"white" if entities[guid]["entityType"] == "dataset" else "aliceblue"}">
    <tr>
        <td rowspan="2" fixedsize="true" width="20" height="20"><img src="{icon_dir + entities[guid]["icon"]}" scale="true"/></td>
        <td align="left" colspan="2" border="0"><b>{entities[guid]["name"]}</b></td>
    </tr>
    <tr>
        <td colspan="2" align="left"><b><font point-size="10">{entities[guid]["typeName"]}</font></b></td>
    </tr>
    {entityInfo}
    <tr><td colspan="2">{classifications}</td></tr>
    {columns}
</table>
        >'''
    )

for relation in response["relations"]:
    lineage04.edge(relation["fromEntityId"], relation["toEntityId"])

lineage04.render("lineage/04.gv")
SVG("lineage/04.gv.svg")

結果

こんな感じになりました。
多少は見やすくなったでしょうか。。エンティティのtypeName毎に設定やアイコンを変えたりと汎用性のない作りにはなってしまいましたが、Lineageの出力自体はできました。