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


# はじめに
[Azure Purview](https://azure.microsoft.com/services/purview/) は、Microsoft が 2020 年に発表した、統合データガバナンスサービスです。日本語ではパァビュー(pˈɚːvjuː)という感じで発音します。プレビューではないです。

Azure Purview を使えば、オンプレでもマルチクラウドでも SaaS でも対応しているデータソースであればメタデータカタログとして集中管理することができるようになります。すでに日本の企業でも導入が始まっており、組織内でデータ利活用を始める際のファーストステップとしてデータの場所をカタログ化し、検索可能にすることの重要性が認知され始めています。また、新たにデータ分析基盤を構築する際の基本的コンポーネントとして取り入れられています。

# Azure Purview 開発
Azure Purview は オープンソースのコア基盤ガバナンスサービス である、[Apache Atlas](https://atlas.apache.org/#/) Open API エコシステムをベースとしており、Microsoft によっていくつかの機能拡張と追加が行われています。

Azure Purview では Atlas と同様、管理するメタデータオブジェクトのモデルを定義できます。考え方はオブジェクト指向プログラミングととても良く似ています。この辺の基礎的な部分は Atlas の [Type System](https://atlas.apache.org/1.2.0/TypeSystem.html) を参照してください。そして、Apache Atlas API 互換の [REST API](https://docs.microsoft.com/rest/api/purview/) や Python クライアント([PyApacheAtlas](https://github.com/wjohnson/pyapacheatlas))を使用してあらゆる操作が可能になっています。

今回は PyApacheAtlas を使って遊んでみたいと思います。

# 事前準備
[こちら](https://docs.microsoft.com/azure/purview/tutorial-using-rest-apis)を参照し、以下を完了してください。クライアントの PyApacheAtlas から Azure Purview を操作するために必要です。

1. Azure Purview アカウント作成
1. サービス プリンシパル (アプリケーション) を作成
1. サービス プリンシパルを使用して認証を設定する
1. トークンを取得する
1. PyApacheAtlas のインストール (`pip install pyapacheatlas`)

## Azure Purview への接続
```python
import json
import os
from pyapacheatlas.auth import ServicePrincipalAuthentication
from pyapacheatlas.core import PurviewClient

SUBSCRIPTION_ID = "" # fill in
RESOURCE_GROUP = "" # fill in
PURVIEW_NAME = "" # fill in
SERVICE_PRINCIPAL_NAME = "" # fill in

oauth = ServicePrincipalAuthentication(
    tenant_id=os.environ.get("TENANT_ID", "<Service Principal Tenant ID>"),
    client_id=os.environ.get("CLIENT_ID", "<Service Principal Client ID>"),
    client_secret=os.environ.get("CLIENT_SECRET","<Service Principal Client Secret>")
)

# PurviewClient のインスタンス化
client = PurviewClient(
    account_name = os.environ.get("PURVIEW_NAME", PURVIEW_NAME),
    authentication=oauth
)
```

## エンティティの作成
まずは以下のコードを打ち込んでみましょう。

```python
from pyapacheatlas.core import AtlasEntity

input01_qn = "pyapacheatlas://hanainput01"
dataset_type_name = "DataSet"

input01 = AtlasEntity(
    name="hanainput01",
    typeName=dataset_type_name,
    qualified_name=input01_qn,
    guid="-100"
)

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

何が起きたかを確認するには、Azure Purview Studio をブラウザで開き、Data map->Collenctions->[Purview 名]をウォッチします。Atlas API で登録した場合、Scan しなくてもすぐに結果を確認できます。Python コードを実行後、 Refresh ボタンをクリックすると、Assets の値が 1 つ増えるかと思います。

![001.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/33982c98-6273-f790-9f4c-922570be678e.jpeg)

Assets のボックスをクリックすると、Assets 一覧が表示され、そこに今回作成した `hanainput01` エンティティが出てきます。

![002.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/91ab1537-5ff5-daa1-c9b1-09d335b46df5.png)

これをクリックすると、最もシンプルなエンティティのメタ情報が表示されます。

![003.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/f6436bf7-ab50-c98e-ac92-28dc718f56d9.png)

これを `Hello, world!` としましょう。

## 型
さて、最初のエンティティ作成では、`typeName` を `DataSet` と指定しましたが、これはどういう意味でしょうか。Atlas で扱うエンティティには必ず型の定義が必要で、型には事前定義された型とそれ継承する型が多数存在します。

例えば、以下のコードを実行します。

```python
from pyapacheatlas.core import AtlasEntity

input01_qn = "pyapacheatlas://hanainput01"
dataset_type_name = "azure_datalake_gen2_filesystem"

input01 = AtlasEntity(
    name="hanainput01",
    typeName=dataset_type_name,
    qualified_name=input01_qn,
    guid="-100"
)

results = client.upload_entities(
    batch=[input01]
)

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

`typeName` を `azure_datalake_gen2_filesystem` に変更し、その他の値は変更しません。

![004.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/846b8f49-a64b-4446-2ccb-9958a436a8fc.png)

表示名が同じで、`Source type` が `Azure Data Lake Storage Gen2` となっているエンティティが新たに作成されました。

![005.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/a34f6cd0-dc08-8045-efeb-54e18b4a42e0.png)

アイコンも変わってますね。型を使えば、Azure Purview 上でデータのソースの種類を出しわけることができそうです。では、Azure Purview 上で定義されている型の一覧を取得するにはどうすればいいかというと、`get_all_typedefs()` メソッドを使用します。

```python
all_typedefs = client.get_all_typedefs()

with open("get_all_typedefs.json", "w") as fp:
    json.dump(all_typedefs, fp)
```

はい、なんと 2MB もの JSON ファイルが生成されました…　読むのつれぇわ…
これを丹念に JSON Editor などで見ていくと Azure Purview の全体像が見えてきます。

型の継承関係は以下のようになっており、DataSet 型の下に様々なデータソースが事前定義されていることが分かります。データベースの場合、テーブルとカラムというように分かれている点が重要です。Atlas ではカラム 1 つが 1 つのエンティティとして扱われます。

```
Referenceable
  └ Asset
    └ DataSet
       ├ azure_datalake_gen2_filesystem
       ├ azure_datalake_gen2_path
       ├ azure_synapse_dedicated_sql_table
       ├ azure_synapse_dedicated_sql_column
       ├ tabular_schema
        ...187 items
```

ちなみに、型名が分かっている場合は、`get_typedef` メソッドを使用することでピンポイントで定義を取得できます。

```python
# azure_datalake_gen2_filesystem の型定義
typedef= client.get_typedef(name="azure_datalake_gen2_filesystem")
print(json.dumps(typedef, indent=2))

# DataSet の型定義
typedef= client.get_typedef(name="DataSet")
print(json.dumps(typedef, indent=2))

```

DataSet 型を継承している型は、属性としてスキーマ情報を持ったりデータ変換情報を保持することで系列(Lineage)グラフを Azure Purview 上で表示できるようになります。

基本的に Azure Purview で[対応しているデータソース](https://docs.microsoft.com/azure/purview/purview-connector-overview)は、スキャンでこれらのエンティティが**自動作成**されますが、型を自分で定義して、カスタムデータソースを作成することができるので、ご自分のソリューションを Azure Purview と統合することができます。

## 属性の追加
次に、メタデータ管理の中心となる属性の付加を行います。
以下のコードを実行します。

```python
from pyapacheatlas.core import AtlasEntity

input01_qn = "pyapacheatlas://hanainput01"
dataset_type_name = "azure_datalake_gen2_filesystem"

input01 = AtlasEntity(
    name="hanainput01",
    typeName=dataset_type_name,
    qualified_name=input01_qn,
    guid="-100",
    attributes = {
        "description": "ファイルについての説明"
}
)

results = client.upload_entities(
    batch=[input01]
)

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

![006.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/6d7e0a35-9644-33e7-16f5-13780a8daf23.png)

Asset description 項目に値が表示されましたね！
このように、エンティティに属性を付加するには、`attributes` プロパティに辞書型で属性を指定します。指定できる属性は、型の中の `attributeDefs` で定義されていなければなりません。今回追加した `description` は 上位の Asset 型の中で定義されています。

コード実行時、result の中で `"UPDATE"` と出力されたかと思いますが、`upload_entities` メソッド実行時、`qualified_name` と `typeName` がキーとなり、すでに存在している場合は、上書きされます。`qualified_name` にはデータセットが一意の値になるように命名します。例：`https://purviewhanatest1adcadls.dfs.core.windows.net/starter1/{Year}/{Month}/{Day}/Contoso_AccountsPayable_{N}.csv`

## 連絡先の追加
エンティティの Contacts タブでは、データセットへの連絡先を設定することができます。

```python
input01 = AtlasEntity(
    name="hanainput01",
    typeName="azure_datalake_gen2_filesystem",
    qualified_name="pyapacheatlas://hanainput01",
    guid="-100",
    contacts= { "Expert": [],
            "Owner": [
              {
                "id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                "info": "データを使いたかったら連絡して😎"
              }
            ]
          }
)
results = client.upload_entities(
    batch=[input01]
)
print(json.dumps(results, indent=2))
```

![022.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/a5ee4e18-c78c-4772-e95a-75d94c4b140c.png)

はい、このようにデータセットのオーナーが追加されましたね。`id` には、Microsoft Graph オブジェクトの ID を設定します。連絡先を設定しておくことで、データ利用者がデータに関する質問や使用可否を問い合わせることができるようになります。


## エンティティを操作する
### エンティティを取得する

Atlas ではエンティティを一意に特定する値として GUID もしくは qualifiedName と typeName の組み合わせが使用されます。GUID は、Azure Purview Studio で特定のエンティティを開いたとき、URL に含まれている guid と同一です。

```python
# qualifiedName と typeName からエンティティを取得
response = client.get_entity(qualifiedName="pyapacheatlas://hanainput01", typeName="azure_datalake_gen2_filesystem")
print(json.dumps(response, indent=2))

# GUID からエンティティを取得
response = client.get_entity(guid="41df8786-c2da-4384-8397-f93ef52a5d0b")
print(json.dumps(response, indent=2))
```

`get_entity` メソッドの戻り値は dict 型なので、これを pyapacheatlas.core.entity の AtlasEntity クラスに変換するには、`from_json` メソッドを使用します。後から既存のエンティティをロードして使いたい場合などに使用します。

```python
AtlasEntity.from_json(response['entities'][0])
```

### エンティティを検索する
エンティティを検索するには、 `search_entities` メソッドを使用します。ここでは[検索クエリ構文](https://docs.microsoft.com/azure/purview/how-to-search-catalog#search-query-syntax)が使用できます。

```python
search = client.search_entities("hana*")
for entity in search:
    print(json.dumps(entity, indent=2))
```

### エンティティを削除する
エンティティを削除するには、`delete_entity` メソッドを使用します。

```python
response = client.delete_entity(guid="6026ecde-4759-4daf-9b95-05a4229f485c")
print(json.dumps(response, indent=2))
```

## スキーマの追加
データソースがデータベース系の場合、カラム情報もメタデータとして管理したいですよね。その場合、以下のように実行します。

```python
# 1 つのテーブルを追加
table = AtlasEntity("hanasqldw001","azure_sql_dw_table", "pyapacheatlas://hanasqldw001", guid="-1")
# 2 つのカラムを追加
c1 = AtlasEntity("column1", "azure_sql_dw_column", "pyapacheatlas://hanasqldw001#col1", guid="-2", attributes={"data_type":"str"})
c2 = AtlasEntity("column2", "azure_sql_dw_column", "pyapacheatlas://hanasqldw001#col2", guid="-3", attributes={"data_type":"int"})

columns_to_add = [c1, c2]
table.addRelationship(columns=[c.to_json(minimum=True) for c in columns_to_add])
result = client.upload_entities([table, c1, c2])

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

![008.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/1bb3d9bd-f6a7-c0f0-b9fe-37de16ec553b.png)

Azure Dedicated SQL Pool Table 型のテーブルを作ってその中に含まれるカラムを Azure Dedicated SQL Pool Column 型で作成しました。Atlas では 1 カラム 1 エンティティとして扱います。そして、`addRelationship` メソッドを使用して、**関係** を作成しました。これによってエンティティ間の関係性を表現することができるようになります。

CSV ファイルなんかもヘッダーがあるので、同様にスキーマを追加することができます。この場合 `delimited_text_schema` 型のエンティティに `delimited_text_schema_columns` 型のカラムエンティティを関連付けます。

## 分類の追加
エンティティに分類を追加するには、以下のように実行します。

```python
from pyapacheatlas.core import AtlasClassification

# 追加したいエンティティの GUID
guid = "5e597602-af21-4b5f-96f3-6cf6f6f60000"

# 1 つのエンティティを複数のカテゴリーに分類
one_entity_multi_class = client.classify_entity(
    guid=guid, 
    classifications=[
        AtlasClassification("MICROSOFT.PERSONAL.JAPANESE.MY.NUMBER.CORPORATE"),
        AtlasClassification("MICROSOFT.PERSONAL.NAME")
        ],
    force_update=True
)
print(json.dumps(one_entity_multi_class, indent=2))
```

![009.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/674344/5366f89f-ff83-d1d6-4f95-fa4e53e5502f.png)

おー分類が追加されましたね。
この分類の事前定義については、`get_all_typedefs()` の `root/classificationDefs` 以下に定義されています。

特定の分類を削除するには、`declassify_entity` メソッドで削除できます。

```python
client.declassify_entity(guid=guid,classificationName="MICROSOFT.PERSONAL.JAPANESE.MY.NUMBER.CORPORATE")
```

さて、基本的な操作について覚えたところで、次はもっと面白いデータ変換やリネージュと呼ばれるデータ系列の作成をしていきたいと思います。が、記事が長くなってきたので、次回の記事にて書こうと思います。

こんな感じで遊びながら触ってるだけでも内部の仕組みについて理解できると思いますので、皆さんも [Azure Purview](https://azure.microsoft.com/services/purview/) を触ってみてほしいなと思います。そして、一緒に新しいソリューションを作っていけたら幸いです。

# つづきはこちら
https://qiita.com/nohanaga/items/e053ed79a5aff398b2d5


# 参考
https://github.com/wjohnson/pyapacheatlas

pyapacheatlas の Documents は doc 以下で Sphinx をビルドしてください。
samples の中にたくさんの使えるコードがあります。