Azure Data Factory or Synapse Mapping Dataflowの変更フィードの紹介

## はじめに

[Flowlets and Change Feed now GA in Azure Data Factory](https://techcommunity.microsoft.com/t5/azure-data-factory-blog/flowlets-and-change-feed-now-ga-in-azure-data-factory/ba-p/3267450)

の2機能がGAしましたので、実用タイミングも踏まえて試してみます。


## 前提知識

[What's Data Lake ? Azure Data Lake best practice](https://speakerdeck.com/ryomaru0825/whats-data-lake-azure-data-lake-best-practice)

より、データレイクの層構造がわかるとユースケースが想像しやすいと思います。
※今回はrawにDelta Lakeなどを利用しない構成で考えてみてます。sparkが利用できる環境ではrawにはDelta Lakeを用いることをお勧めします。


## ユースケース

以下のようなraw->entichのケースで使ってみます。

顧客マスタが毎日日付の入ったcsvで連携され、以下のゾーンを利用して統合されます。

raw:連携されたファイルをそのまま蓄積
enrich:rawに連携された **新規ファイル** のみをdelta lakeとして変換し、upsert

## 手順

### 0. 環境とデータを準備

1 Synapse Analyticsをデプロイします。（ADFでもいいですが、datalakeと接続の作成が必要です）

[Synapse ワークスペースの作成](https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-create-workspace)

2 データを準備します。

[Azure Data Factory の Mapping Data Flow で CSV ファイルの重複行を削除する](https://qiita.com/nakazax/items/94ffeaf2c9d7ada96cc2)から参考にさせていただきました。

二日分のデータでID=1のレコードのみ更新がかかっています。

ファイル名：customer_20220301.csv

```

"ID","Email","Name"
"1","yamada.hanako@example.com","Yamada Hanako"
"2","tanaka.taro@example.com","Tanaka Taro"

```


ファイル名：customer_20220302.csv

```

"ID","Email","Name"
"1","tanaka.hanako@example.com","Tanaka Hanako"
"2","tanaka.taro@example.com","Tanaka Taro"

```

### 1. Mapping Data flowを作成します。

1 ソースを追加します。

![2022-03-31-14-12-52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/2376d8ce-26b8-e12b-014a-c37e4699e787.png)

Data Lake Strage Gen2、Delimited textのデータセットを作成し、
以下のプロパティに設定します。

パスは **raw/cf_customer** とします。

![2022-03-31-14-14-54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/30a30fa8-6f60-07f6-6565-ea8564dc5669.png)

サンプルファイルからスキーマのインポートを行いましょう。

![2022-03-31-14-15-20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9dc41e11-86d2-a563-39d9-85a15f83785f.png)

2 変更データキャプチャの設定をオンにします。また、ファイル名も格納しておきましょう。この設定により、今後rawゾーンに置かれたファイルのうち、未処理のファイルのみが処理されます。

![2022-03-31-14-16-24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ac33c00c-6326-7442-addb-99ce34dfb537.png)


3 シンクを設定します。

Delta Lake は差分と表示されていますが、気にせず利用します。。。

![2022-03-31-14-17-45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/3c9d2ef1-8540-5396-0fca-64ba9154a406.png)


パスは　**enrich/cf_customer** とします。

![2022-03-31-14-19-43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/f0b21e23-aaad-e430-1dd4-c82d6a9fb16c.png)


4 データ追加時のアクションを設定します。

IDをキーにしてアップサートしましょう。これで最新の状態にマスタが更新されていきます。


![2022-03-31-14-20-42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/fc96cedd-94f4-c4ff-d593-aaa04f06fb2f.png)

キーカラムも忘れずに

![2022-03-31-14-22-22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/572edb5f-7d22-a0f4-3e27-7bb166d59ce2.png)

5 案内にしたがい、**行の変更の追加** をクリック後、すべての行がアップサート対象となるようにtrue()を設定します。

![2022-03-31-14-24-19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a09e0e4d-67cc-d57d-feb4-af17574c9f70.png)


6 パイプラインを構成します。

![2022-03-31-14-23-01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/f517c1ca-d8c0-23c7-d960-1dd7cb5e8d08.png)

### 2. データを配置し、初回の結果を確認します。

1 データをアップします。

![2022-03-31-14-23-44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e79cfb78-7532-e5a8-094d-11cd6b4a9d48.png)

2 デバッグ実行します。

![2022-03-31-14-30-47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/6dc93696-a8c1-3bf9-644e-e347fc390a1e.png)

3 Delta Lakeで対象フォルダをクエリして確認します。

![2022-03-31-14-32-19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/becd63eb-1712-4bd4-9360-24361e02d30e.png)



### 3. 差分データを配置し、2回目の結果を確認します。

1 データをアップします。

![2022-03-31-14-32-43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/2a11c9ad-8a68-807f-6636-3e2b052d36b6.png)

2 デバッグ実行し、処理された行数を確認します。1ファイル分が処理されていることがわかります。


![2022-03-31-14-33-49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/390bd6f3-1ea1-764f-3438-34dc4725c60f.png)


3 正常にデータが更新されました

![2022-03-31-14-36-17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a9380aa4-b591-dfb1-03ab-18bc9748c2f1.png)


## 補足

今回はアップされたデータすべて更新されるような方式ですが、連携データを更新分のみにするか、更新チェックをデータフローで実装すれば更新のコストを削減できると思います。


## 参考

https://docs.microsoft.com/en-us/azure/data-factory/connector-azure-data-lake-storage?tabs=data-factory#change-data-capture

https://www.youtube.com/watch?v=Y9J5J2SRt5k

https://docs.microsoft.com/ja-jp/azure/data-factory/connector-azure-data-lake-storage?tabs=data-factory#change-data-capture-preview

以下の記載があるため、パイプラインの変更には注意

>最後の実行からチェックポイントを常に記録して、そこから変更を取得できるよう、パイプラインとアクティビティ名は変更しないでください。 パイプライン名またはアクティビティ名を変更すると、チェックポイントがリセットされ、次の実行は最初から開始されます。