Azure Purview を遊びながら開発しよう（応用編）

#はじめに
この記事は、[Azure Purview を遊びながら開発しよう（基本編）](https://qiita.com/nohanaga/items/b6f6ba045c1f78a15073)の第二弾となります。

https://qiita.com/nohanaga/items/b6f6ba045c1f78a15073

#データ変換
Azure Purview では、データの変換操作をエンティティとして保持することができます。そのためのエンティティとして `Process` 型が用意されています。

ひとまず以下のコードを実行します。

```python
from pyapacheatlas.core import AtlasProcess

input01 = AtlasEntity("inputprocess01.csv","DataSet", "pyapacheatlas://inputprocess01", guid=-100)
output01= AtlasEntity("outputprocess01.csv","DataSet", "pyapacheatlas://outputprocess01", guid=-101)

process = AtlasProcess(
    name="データ変換処理1",
    typeName="Process",
    qualified_name="pyapacheatlas://hanaprocess1",
    inputs=[input01],
    outputs=[output01],
    guid=-102
)

results = client.upload_entities(
    batch=[input01, output01, process]
)

print(json.dumps(results, indent=2))
```

そして、"データ変換" などのキーワードでカタログを検索すると、`Process` 型のエンティティがヒットします。

![011.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/ddf6dd89-5a65-3e86-741f-fcd82c1d92ec.png)


pyapacheatlas.core の AtlasProcess クラスに inputs と outputs となるエンティティを指定してあげることで、データの処理過程を[系列(Lineage)](https://docs.microsoft.com/azure/purview/catalog-lineage-user-guide)として表現することができます。Azure Purview では、[Azure Data Factory](https://docs.microsoft.com/azure/purview/how-to-link-azure-data-factory) や [Azure Synapse](https://docs.microsoft.com/azure/purview/how-to-lineage-azure-synapse-analytics) のコピーアクティビティやデータフローアクティビティなどのデータ処理が系列として表示できます。これを自動的に連携するには事前にそれぞれのサービス側から Azure Purview へ[接続しておく](https://docs.microsoft.com/azure/data-factory/connect-data-factory-to-azure-purview)必要があります。

それでは、Azure Purview で定義されている `Process` 型についてもう少し詳しく調べてみましょう。

```python
typedef = client.get_typedef(name="Process")
print(json.dumps(typedef, indent=2))
```

出力結果から、`Asset` 型を継承していることが分かります。つまり、`Process` 型は `DataSet`型と同階層に属する型ということです。

```
Referenceable
  └ Asset
    ├ DataSet
    └ Process
       ├ azure_synapse_activity
       ├ azure_synapse_pipeline
       ├ adf_process
       ├ azure_sql_stored_procedure
       ├ azure_sql_query_run
        ...46 items
```

ただし、それぞれの型の下にもサブタイプが存在し、結構複雑です。しっかり型定義を見ていく必要があります。例えば Azure Data Factory で Copy アクティビティを実行した場合、`adf_copy_activity` が作成されます。この型には、`adf_pipeline` 型のエンティティを必要としますし、`adf_copy_operation` 型のエンティティをオプションで受け入れます。


## 複数データソースを系列に追加する
ETL/ELT 処理では複数のデータソースを入力し、一つの出力を得ることもあるかと思いますがそれも簡単にできます。AtlasEntity を作成して、AtlasProcess の inputs 配列に追加するだけです。

```python
from pyapacheatlas.core import AtlasProcess

input01 = AtlasEntity("inputprocess01.csv","DataSet", "pyapacheatlas://inputprocess01", guid=-100)
input02 = AtlasEntity("inputprocess02.csv","DataSet", "pyapacheatlas://inputprocess02", guid=-101)
output01= AtlasEntity("outputprocess01.csv","DataSet", "pyapacheatlas://outputprocess01", guid=-102)

process = AtlasProcess(
    name="データ変換処理1",
    typeName="Process",
    qualified_name="pyapacheatlas://hanaprocess1",
    inputs=[input01, input02],
    outputs=[output01],
    guid=-103
)

results = client.upload_entities(
    batch=[input01, input02, output01, process]
)

print(json.dumps(results, indent=2))
```

![012.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/69c67f60-f764-beb2-5137-e3da49314f0a.png)

はい、ファイルが追加されましたね。もし、`typeName` に `azure_synapse_pipeline` に変えて実行すると…

![013.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/ff60167e-76a3-9dc3-0d4c-7887dc237b29.png)

こんな感じに変わります。ただ、属性や関連に何も値を指定していないので、クリックしても Azure Synapse の該当ページへのリンクも表示されません。

## Azure Data Factory の Copy アクティビティの作成
通常、Azure Purview と正しく接続できていれば Copy アクティビティ実行後、Azure Purview 側で Scan することで系列が自動的に連携されます。でも、内部はどうなっているのか知りたいですよね…　この Copy アクティビティは Atlas API から作成しようとするとかなり複雑ですが、以下のコードを実行すれば一発で作成することができます。

```python
# qualified_name のパスを指定しておく
adf_qualifiedPath = "/subscriptions/12345/resourceGroups/purviewhanatest1/providers/Microsoft.DataFactory/factories"
adf_qName = adf_qualifiedPath + "/adf01"
adf_pipeline_qName = adf_qName + "/pipelines/hanaadf_pipeline1"
adf_activity_qName = adf_pipeline_qName + "/activities/hanaadf_copy_activity1"
adf_operation_qName = adf_activity_qName + "#hanaadf_copy_operation1"
# 最上位は azure_data_factory 型
adf = AtlasEntity("adf01","azure_data_factory", adf_qName, guid=-100)
# Process 型への input と output
input01 = AtlasEntity("sql_input01","azure_sql_dw_table", "pyapacheatlas://sql_input01", guid=-101)
output01= AtlasEntity("sql_output01","azure_sql_table", "pyapacheatlas://sql_output01", guid=-102)
# Pipeline の作成
proc_pipeline = AtlasProcess(
    name="データ変換処理_パイプライン",
    typeName="adf_pipeline",
    qualified_name=adf_pipeline_qName,
    guid=-103,
    inputs=[],
    outputs=[]
)
# Activity の作成
proc_activity = AtlasProcess(
    name="データ変換処理_Copyアクティビティ",
    typeName="adf_copy_activity",
    qualified_name=adf_activity_qName,
    inputs=[],
    outputs=[],
    guid=-104,
    attributes = {"status": "Completed"}
)
# Operation の作成
proc_operation = AtlasProcess(
    name="データ変換処理_オペレーション",
    typeName="adf_copy_operation",
    qualified_name=adf_operation_qName,
    guid=-105,
    inputs=[input01],
    outputs=[output01],
)
# Relationship の作成
proc_activity.addRelationship(subProcesses=[proc_operation.to_json(minimum=True)])
proc_pipeline.addRelationship(subProcesses=[proc_activity.to_json(minimum=True)])
adf.addRelationship(pipelines=[proc_pipeline.to_json(minimum=True)])

results = client.upload_entities(
    batch=[adf, proc_pipeline, proc_activity, proc_operation, input01, output01]
)
print(json.dumps(results, indent=2))
```

![015.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/03a6bc2b-5170-24d0-70f9-ac61f14c366a.png)


各 Process 型に特有の設定がありますが、Azure Data Factory の Copy アクティビティの例を知っておけば他にも応用が利くと思います。単に `adf_copy_activity` 型のエンティティに input/output を指定するだけでは正しく系列が表示されません。この場合は、`adf_copy_activity` の下に `adf_copy_operation` 型のエンティティを作成し、そちらに指定してあげる必要があります。そして、上位のエンティティとの間に**関係**を作成します。

さらに、正しい Copy アクティビティはこのように階層が表示されます。

![016.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/574afd5e-a057-b047-831f-d38853b1aba9.png)

この階層表示および Related タブの階層は、関係を作成しただけでは表示されません。既定のパス形式に沿って設定された `qualified_name` が必要です。`adf_copy_activity` の場合は上記コードのようなパスが必要ですが、カスタムする場合でも `qualified_name` の命名規則はしっかり決めておいた方がいいです。

ちなみに、系列ボックス上の「Open in Azure Data Factory」ボタンで開かれるリンク先 URL の生成に `qualified_name` が使用されます。

## 系列のカラムマッピング作成
これまではデータセット粒度での系列でしたが、Azure Purview ではさらにカラム粒度での系列を表示し、データ変換処理を経て、どのようにカラムが伝播していったのかを追跡することができます。

```python
from pyapacheatlas.core import AtlasProcess
from pyapacheatlas.core.typedef import EntityTypeDef, AtlasAttributeDef

# columnMapping 属性の定義を含むカスタムプロセスエンティティ型を定義
procType = EntityTypeDef(
    "ProcessWithColumnMapping",
    superTypes=["Process"],
    attributeDefs = [
        AtlasAttributeDef("columnMapping")
    ]
)
# 型定義のアップロード
type_results = client.upload_typedefs(entityDefs=[procType], force_update=True)
print(json.dumps(type_results,indent=2))

input01 = AtlasEntity("sql_col_input01","azure_sql_dw_table", "pyapacheatlas://sql_col_input01", guid=-100)
output01= AtlasEntity("sql_col_output01","azure_sql_table", "pyapacheatlas://sql_col_output01", guid=-101)
# カラムマッピングの定義
column_mapping = [
    {"ColumnMapping": [
        {"Source": "In01_ID", "Sink": "Out01_ID"},
        {"Source": "In01_Name", "Sink": "Out01_Name"}],
        "DatasetMapping": {
        "Source": input01.qualifiedName, "Sink": output01.qualifiedName}
     }
]

process = AtlasProcess(
    name="データ変換処理_カラムマッピング",
    typeName="ProcessWithColumnMapping",
    qualified_name="pyapacheatlas://hana_col_process1",
    inputs=[input01],
    outputs=[output01],
    guid=-102,
    attributes={"columnMapping": json.dumps(column_mapping)}
)

results = client.upload_entities(
    batch=[input01, output01, process]
)

print(json.dumps(results, indent=2))
```

![017.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/57248a34-4501-b18f-8fb9-223b8167a60e.png)

このように入力テーブルのカラムがデータ変換処理の後、どこへ伝播したかを示す系列を作成できました。この時、`EntityTypeDef` クラスを使って `columnMapping` 属性を持つ新たな Process 型のエンティティ `ProcessWithColumnMapping` を定義しました。これは既定でカラムマッピング属性を保持できるシンプルな Process 型が無かったためです。`azure_synapse_operation` 型や `adf_copy_activity` 型、`ssis_package_process` 型などは既定でカラムマッピング属性を保持できますが、前述のとおり作るのがめんどくさいです。

## 複数のカラムマッピング作成
データ変換処理で 1 つのカラムから複数のカラムを生成するパターンもよくあると思います。これも簡単にできます。

```python
column_mapping = [
    {"ColumnMapping": [
        {"Source": "In01_ID", "Sink": "Out01_ID"},
        {"Source": "In01_Name", "Sink": "Out01_LastName"},
        {"Source": "In01_Name", "Sink": "Out01_FirstName"}],
        "DatasetMapping": {
        "Source": input01.qualifiedName, "Sink": output01.qualifiedName}
     }
]

process = AtlasProcess(
    name="データ変換処理_カラムマッピング",
    typeName="ProcessWithColumnMapping",
    qualified_name="pyapacheatlas://hana_col_process1",
    inputs=[input01],
    outputs=[output01],
    guid=-102,
    attributes={"columnMapping": json.dumps(column_mapping)}
)

results = client.upload_entities(
    batch=[input01, output01, process]
)

print(json.dumps(results, indent=2))
```

![018.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/d65ac317-5d72-2849-1a93-03a12ed0ae38.png)

いいですね！
試しにさっき作った Azure Data Factory の Copy アクティビティにカラムマッピングを追加するにはどうすればいいかというと、

```python
input01 = AtlasEntity("sql_input01","azure_sql_dw_table", "pyapacheatlas://sql_input01", guid=-101)
output01= AtlasEntity("sql_output01","azure_sql_table", "pyapacheatlas://sql_output01", guid=-102)

column_mapping = [
    {"ColumnMapping": [
        {"Source": "In01_ID", "Sink": "Out01_ID"},
        {"Source": "In01_Name", "Sink": "Out01_Name"}],
        "DatasetMapping": {
        "Source": input01.qualifiedName, "Sink": output01.qualifiedName}
     }
]

proc_activity = AtlasProcess(
    name="データ変換処理_Copyアクティビティ",
    typeName="adf_copy_activity",
    qualified_name=ADF_QUALIFIED_NAME + "/adf01/pipelines/hanaadf_pipeline1/activities/hanaadf_copy_activity1",
    inputs=[],
    outputs=[],
    guid=-103,
    attributes = {"status": "Completed", "columnMapping": json.dumps(column_mapping)}
)

results = client.upload_entities(
    batch=[proc_activity]
)
print(json.dumps(results, indent=2))
```

![019.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/686a679b-765c-e488-4229-9103ba018c51.png)

サクッと行けましたね！
Process 型に `columnMapping` という属性が無いと追加できないということです。

## さらに次のデータ処理を追加する
前回の出力データに対して、さらに後段のデータ処理を追加するには、以下のように書きます。

```python
from pyapacheatlas.core import AtlasEntity, AtlasProcess

output01= AtlasEntity("sql_output01","azure_sql_table", "pyapacheatlas://sql_output01", guid=-100)
next01  = AtlasEntity("sql_next01","azure_sql_table", "pyapacheatlas://sql_next01", guid=-101)

process = AtlasProcess(
    name="データ変換処理その2",
    typeName="Process",
    qualified_name= "pyapacheatlas://hanaprocess2",
    inputs=[output01],
    outputs=[next01],
    guid=-102
)

results = client.upload_entities(
    batch=[output01, next01, process]
)

print(json.dumps(results, indent=2))
```
![020.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/72bad789-e6c8-f6e0-0a02-54ab3f524637.png)

簡単にできましたね。 AtlasProcess の inputs に前回の AtlasEntity を指定するだけです。

## 関係を作成する別の方法
すでに作成済みのエンティティ同士を関連付けるには、`upload_relationship` メソッドを使用することもできます。

```python
relationship01 = client.get_entity(qualifiedName="pyapacheatlas://sql_output01", typeName="azure_sql_table")
relationship02 = client.get_entity(qualifiedName="pyapacheatlas://hanaprocess3", typeName="Process")

relationship = {
    "typeName": "dataset_process_inputs",
    "attributes": {},
    "guid": -100,
    "end1": {
        "guid": relationship01['entities'][0]['guid']
    },
    "end2": {
        "guid": relationship02['entities'][0]['guid']
    }
}

# 新しい関係をアップロードする
results = client.upload_relationship(relationship)
print(json.dumps(results, indent=2))
```

![021.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/c6e8cff5-cd08-7b02-825c-c4b1c7266c25.png)

既定で作成できる関係は、データセットとデータ変換やテーブルとカラムなどの定義された関係に限られます。この関係の事前定義については、`get_all_typedefs()` の `root/relationshipDefs` 以下に定義されています。データ変換とインプット側の関係を作成する際は `dataset_process_inputs` 型を、データ変換とアウトプット側の関係を作成する場合は、`process_dataset_outputs` 型を使用します。

# さいごに
はい、こんな感じで Azure Purview の開発を楽しんでいただければと思います。Azure Purview にはこの他にも色々な機能があるので試してみるのも面白いのではないでしょうか。そして Azure Purview の開発を通して、新たなデータガバナンスエコシステムが活性化されることを楽しみにしています。