OpenMetadata の Profiler について調べてみた

この記事は [気にはなってるけど触ってないビッグデータ系のツール・サービスを触る Advent Calendar 2022](https://qiita.com/advent-calendar/2022/bigdata-tools-you-heard-not-used) の20日目の記事です。

# OpenMetadata とは
[OpenMetadata](https://open-metadata.org/) は OSS のデータカタログツールです。最近、SaaS 版の [Collate](https://www.getcollate.io/) も登場しました。OpenMetadata では、主にメタデータの管理、テーブルレベルのデータリネージやデータプロファイルなどの機能を持っています。

機能の概要については Classmethod さんの記事がわかりやすいです
https://dev.classmethod.jp/articles/introduction-of-open-metadata/

# なぜデータカタログが必要か
まず、データカタログは分析基盤の補助システムという位置付けです。
分析基盤の利用者が困ることとして、

- 欲しいデータがどこにあるのかわからない
- データがいつ更新されるのかわからない
- カラムの意味がわからない
- データの分布が知りたい
- データがおかしいときにどこに問い合わせれば良いかわからない

などなど、分析の前に様々なハードルが存在します。
そうしたときに予めドキュメントが整備してあれば、利用者はとても助かると思います。
また、利用者だけでなく運用者にとっても、分析基盤の守りを固める意味でデータカタログを整備していくことがデータの価値を毀損させないためにも重要だと考えます（得てして分析基盤は勝手に壊れることが多いため）。

# 調査のモチベーション
ずばり、あるテーブルのデータの分布を時系列で知りたい！です。プロダクトのリリース後にデータの分布が変わることはよくあります(想定された変化であったり、バグであったり)。僕の経験から、こういうときは利用者からの問い合わせで発覚することが多かったです。

分析基盤をもっと強固にしたい！利用者から問い合わせを受ける前にデータの分布が変わったことを察知したい！そんな動機から何か使えそうなツールがないかデータカタログの調査を始めました。

そうした時に、OpenMetadata の Profiler が気になったので調べてみることにしました。テーブルのプロファイリングは恐らく様々な SQL が発行されるはずなので、重いテーブルに対して SQL が発行されまくって課金死しないか調べたいと思います(BigQuery を想定しています)。

# OpenMetadata の Architecture
https://docs.open-metadata.org/developers/architecture より引用
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/d750d3f6-1d1a-e6dd-52b8-bdbe1aea2cf6.png)

- データカタログを利用するための Web UI
- API Server(Drop Wizard)
- Web UI からメタデータを検索するための検索エンジン(Elasticsearch)
- 定期的にメタデータを取り込むためのスケジューラー(Airflow)
- メタデータを格納するための DB(MySQL)

# 触ってみる
やること

1. docker-compose.yml を取得する
1. OpenMetadata Web UI にアクセスできることを確認する
1. Airflow Web UI にアクセスできることを確認する
1. OpenMetadata Web UI で BigQuery への接続設定を行う
1. BigQuery にサンプルデータを用意する
1. OpenMetadata Web UI 上から BigQuery のサンプルデータの Metadata を取り込む
1. OpenMetadata Web UI 上から BigQuery のサンプルデータの Profile を行う
1. BigQuery で OpenMetadata からどんなクエリが発行されたか確認する

## docker-compose.yml を取得する
https://github.com/open-metadata/OpenMetadata/releases

のページから docker-compose.yml を取得して

![スクリーンショット 2022-12-18 23.35.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/0932fac3-9f6a-53c6-c6f5-0430b9c66b72.png)

```bash
$ docker-compose up -d
```

を実行します（それなりに時間が掛かります）

続いて OpenMetadata の Web UI と Airflow の Web UI を確認してみます

## OpenMetadata Web UI にアクセスできることを確認する
http://localhost:8585
admin/admin でログインします

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/d8ee29b9-e8b2-3a60-02ce-1a114a8bceb2.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/118824fb-cb4b-852c-a306-64ad7523069c.png)


## Airflow Web UI にアクセスできることを確認する
http://localhost:8080
admin/admin でログインします

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/181a155a-c47a-4448-bad2-1469fc78fa91.png)

初期状態でいくつか Sample DAG が登録されています

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/70959d08-44cb-5083-4dc2-9e2b80da2689.png)

## OpenMetadata Web UI で BigQuery への接続設定を行う

Settings を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/64594236-9828-2593-9f56-1a544527cc69.png)

Databases を押下し、Add new Database Service を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/22c6052d-2eb3-60fa-e017-38fab497c9cd.png)

BigQuery を選択して、画面下部の Next を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/18e85208-bcde-a3e2-3434-4881dee70148.png)

Service Name を入力します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/bbf11f06-337b-9a58-d7db-8bfcb31d7e84.png)

GCP 上でサービスアカウントを作成、クレデンシャルキーを発行して登録します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/6584a82d-e993-acba-163b-6d697864925b.png)

接続テストを行い、接続設定を登録します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/ef7e2683-33ae-7841-a3e2-f7015a924323.png)

## BigQuery にサンプルデータを用意する
サンプルとして、bigquery-public-data の taxi トリップデータを使うことにします

`bigquery-public-data.chicago_taxi_trips.taxi_trips` テーブルを自身のプロジェクトにコピーします

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/bbd7a8d7-87d5-7960-86ee-7b5802eb1994.png)

約74GB なので、フルスキャン数回レベルでは課金死には至らなさそうです

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/965e67b9-157a-fc92-f36c-97d5602bfe8a.png)

## OpenMetadata Web UI 上から BigQuery の Metadata を取り込む

さきほどの OpenMetadata の接続設定の画面から Add Ingestion を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/ef7e2683-33ae-7841-a3e2-f7015a924323.png)

取り込むテーブルのフィルタリングを行います(特に設定しない)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/1be53dbd-aee7-09c5-96a2-cd522ceee89e.png)

dbt の設定ファイルを読み込めるようですが今回はスキップします

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/19cdc5b2-588d-2bb2-d5d8-b0238d83fb53.png)

取り込み間隔の設定です

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/8eb533f0-ddde-dff0-f6e2-78d4f2c77023.png)

今回は手動で実行したいので none を選択して、Add & Deploy を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/0728d194-14d2-d735-aca3-c0acc1de779f.png)

取り込み設定ができました。Run を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/1814f701-edab-9994-6f40-1ce7709ea30e.png)

Airflow の Web UI を確認すると、DAG が追加されています
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/f59d32e2-39a5-ef04-71b9-4c6ecb593578.png)

無事、完了したようです

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/f1bb4ded-f2cb-f846-e53c-c7781139d072.png)

取り込まれたデータを確認します。トップ画面から Tables へ遷移します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/6a5890ac-ded7-a1f8-5ff0-1f910315742d.png)

taxi_trips が取り込まれています

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/b497c02b-4e04-f774-368f-4d01b611ecd7.png)

カラム、型、Description が同期されています

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/61761cfc-f765-c1bf-fc8e-db105e8c9abc.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/2fe2934b-bd76-76c4-b475-a2f7e573204d.png)


## OpenMetadata Web UI 上から BigQuery のサンプルデータの Profile を行う
続いて Profile を行います。Database Services の BigQuery の接続設定から Add Ingestion を行います

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/c3f2c11f-7913-9d06-f021-04cdab0addfc.png)

ここから Add Profiler Ingestion を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/f0ce2c59-ff3c-7ae5-0784-901d96906179.png)

Metadata を取り込んだ時と流れは同じです

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/928d969d-d8a2-5685-61f2-2711913b11d5.png)

Profiler Ingestion のジョブを実行するために、Run を押下します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/4f848fea-30f2-0325-fc04-a3bec1e6797d.png)

同じように Airflow に DAG が追加されています

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/d6972d05-e01b-0d89-75bc-db2e3c3624a6.png)

ジョブが完了したら、taxi_trips の Profiler 画面を確認します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/4567925b-b427-66c5-cdb5-90f76f8a5ca3.png)

Profiler では Null の割合、Unique の割合、Distinct の割合などが確認できます

試しに fare (乗車料金)カラムを見てみます
1回だけの取り込みなのであまり面白くないですが、定点でデータを取り込むとデータの変化を確認することができそうです
（Test を設定できるので、そこから通知ができるのかどうか確認したい。。。時間がなくて確認できず）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/5ed28a93-174a-1559-e889-2442c7dfc77c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/5e3c93b2-725c-80b4-b9c7-5d7db3f29d75.png)

## BigQuery で OpenMetadata からどんなクエリが発行されたか確認する

下記コマンドで確認すると、Profiling のために発行されたクエリは71本でした
```bash
$ bq ls -j -a -n 100
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273970/f93c8a69-7a2b-63a4-84a3-8da5e6bf81b7.png)

これらのクエリを詳しく見ていきたいと思います

テーブルのレコード数の集計

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
SELECT count(*) AS `rowCount`, @`param_1` AS `columnCount`, @`param_2` AS `columnNames` 
FROM `chicago_taxi_trips`.`taxi_trips`
 LIMIT @`param_3`
```

各カラムに対して、count, min, max, null count などの集計

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
SELECT avg(LENGTH(`taxi_id`)) AS `mean`, count(`taxi_id`) AS `valuesCount`, count(DISTINCT `taxi_id`) AS `distinctCount`, NULL AS `anon_1`, min(LENGTH(`taxi_id`)) AS `minLength`, NULL AS `anon__1`, max(LENGTH(`taxi_id`)) AS `maxLength`, SUM(CAST(CASE WHEN (`taxi_id` IS NULL) THEN @`param_1` ELSE @`param_2` END AS NUMERIC)) AS `nullCount`, NULL AS `anon__2`, NULL AS `anon__3` 
FROM `chicago_taxi_trips`.`taxi_trips`
 LIMIT @`param_3`
```

各カラムに対して、ユニーク数のカウント

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
WITH `only_once` AS 
(SELECT count(`unique_key`) AS `count_1` 
FROM `chicago_taxi_trips`.`taxi_trips` GROUP BY `unique_key` 
HAVING count(`unique_key`) = @`count_2`)
 SELECT count(*) AS `uniqueCount` 
FROM `only_once`
 LIMIT @`param_1`
```

何をやっているのか不明です。恐らくカラムの型によって本来発行不要なクエリを発行してそうな感じがします

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
SELECT NULL AS `anon_1` 
FROM `chicago_taxi_trips`.`taxi_trips`
 LIMIT @`param_1`
```

数値型カラムに対して、median の集計

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
SELECT percentile_cont(`trip_seconds` , 0.5) OVER() AS `median` 
FROM `chicago_taxi_trips`.`taxi_trips`
 LIMIT @`param_2`
```

恐らくランダムサンプリング

```sql
/* {"app": "OpenMetadata", "version": "0.13.0.1"} */
WITH `taxi_trips_rnd` AS 
(SELECT `taxi_trips`.`unique_key` AS `unique_key`, `taxi_trips`.`taxi_id` AS `taxi_id`, `taxi_trips`.`trip_start_timestamp` AS `trip_start_timestamp`, `taxi_trips`.`trip_end_timestamp` AS `trip_end_timestamp`, `taxi_trips`.`trip_seconds` AS `trip_seconds`, `taxi_trips`.`trip_miles` AS `trip_miles`, `taxi_trips`.`pickup_census_tract` AS `pickup_census_tract`, `taxi_trips`.`dropoff_census_tract` AS `dropoff_census_tract`, `taxi_trips`.`pickup_community_area` AS `pickup_community_area`, `taxi_trips`.`dropoff_community_area` AS `dropoff_community_area`, `taxi_trips`.`fare` AS `fare`, `taxi_trips`.`tips` AS `tips`, `taxi_trips`.`tolls` AS `tolls`, `taxi_trips`.`extras` AS `extras`, `taxi_trips`.`trip_total` AS `trip_total`, `taxi_trips`.`payment_type` AS `payment_type`, `taxi_trips`.`company` AS `company`, `taxi_trips`.`pickup_latitude` AS `pickup_latitude`, `taxi_trips`.`pickup_longitude` AS `pickup_longitude`, `taxi_trips`.`pickup_location` AS `pickup_location`, `taxi_trips`.`dropoff_latitude` AS `dropoff_latitude`, `taxi_trips`.`dropoff_longitude` AS `dropoff_longitude`, `taxi_trips`.`dropoff_location` AS `dropoff_location`, MOD(CAST(100*RAND() AS INT64), @`param_1`) AS `random` 
FROM `chicago_taxi_trips`.`taxi_trips` )
 SELECT `taxi_trips_rnd`.`unique_key` AS `taxi_trips_rnd_unique_key`, `taxi_trips_rnd`.`taxi_id` AS `taxi_trips_rnd_taxi_id`, `taxi_trips_rnd`.`trip_start_timestamp` AS `taxi_trips_rnd_trip_start_timestamp`, `taxi_trips_rnd`.`trip_end_timestamp` AS `taxi_trips_rnd_trip_end_timestamp`, `taxi_trips_rnd`.`trip_seconds` AS `taxi_trips_rnd_trip_seconds`, `taxi_trips_rnd`.`trip_miles` AS `taxi_trips_rnd_trip_miles`, `taxi_trips_rnd`.`pickup_census_tract` AS `taxi_trips_rnd_pickup_census_tract`, `taxi_trips_rnd`.`dropoff_census_tract` AS `taxi_trips_rnd_dropoff_census_tract`, `taxi_trips_rnd`.`pickup_community_area` AS `taxi_trips_rnd_pickup_community_area`, `taxi_trips_rnd`.`dropoff_community_area` AS `taxi_trips_rnd_dropoff_community_area`, `taxi_trips_rnd`.`fare` AS `taxi_trips_rnd_fare`, `taxi_trips_rnd`.`tips` AS `taxi_trips_rnd_tips`, `taxi_trips_rnd`.`tolls` AS `taxi_trips_rnd_tolls`, `taxi_trips_rnd`.`extras` AS `taxi_trips_rnd_extras`, `taxi_trips_rnd`.`trip_total` AS `taxi_trips_rnd_trip_total`, `taxi_trips_rnd`.`payment_type` AS `taxi_trips_rnd_payment_type`, `taxi_trips_rnd`.`company` AS `taxi_trips_rnd_company`, `taxi_trips_rnd`.`pickup_latitude` AS `taxi_trips_rnd_pickup_latitude`, `taxi_trips_rnd`.`pickup_longitude` AS `taxi_trips_rnd_pickup_longitude`, `taxi_trips_rnd`.`pickup_location` AS `taxi_trips_rnd_pickup_location`, `taxi_trips_rnd`.`dropoff_latitude` AS `taxi_trips_rnd_dropoff_latitude`, `taxi_trips_rnd`.`dropoff_longitude` AS `taxi_trips_rnd_dropoff_longitude`, `taxi_trips_rnd`.`dropoff_location` AS `taxi_trips_rnd_dropoff_location` 
FROM `taxi_trips_rnd`
 LIMIT @`param_2`
```

以上、Profiling のために発行されたクエリでした
これらのクエリを発行するときにテーブルのパーティションキーを設定する方法があるのではないのかと思うのですが、
（そうでないと課金死の問題が解決できない）今後の課題としたいと思います

# まとめ
本記事では OpenMetadata の紹介をさせて頂きました。
当初の目的だった Profilier について残課題はいくつかありますが、
なんとなくどういう動きをするのかイメージがついたのではないかと思います。

docker-compose.yml さえあれば簡単に動かせると思いますので、
皆様も一度 OpenMetadata を触ってみてはいかがでしょうか・