Azure Data Factory のメタデータ駆動のコピータスクにて利用するパイプラインをデータセットごとに共通化する方法

## 概要

Azure Data Factory のメタデータ駆動のコピータスクにて利用するパイプラインをデータセットごとに共通化する方法を共有します。

メタデータ駆動のコピータスクにて作成されるパイプラインでは、`MetadataDrivenCopyTask_xxx_TopLevel`パイプラインにてシステム変数である`pipeline().Pipeline`と`pipeline().TriggerName`が利用されていることで、パイプラインの共通化を実施することができません。本手順では、その２つのシステム変数が利用されている箇所をパラメータにより指定できるようにすることでパイプラインを共通化します。

なお、データソースが異なる場合にはパイプラインを共通化できないことに注意してください。

> ADF では、IR 名、データベースの種類、ファイル形式の種類をパラメーター化できません。 たとえば、Oracle Server と SQL Server の両方からデータを取り込む場合は、2 つの異なるパラメーター化されたパイプラインが必要です。

引用元：[データのコピー ツールでメタデータ駆動型の方法を使用して大規模なデータ コピー パイプラインを作成する - Azure Data Factory | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/data-factory/copy-data-tool-metadata-driven#known-limitations)

## 構築手順

### 1. Azure SQL Database にてテーブル作成と初期データを登録

```sql
IF  OBJECT_ID(N'dbo.adf_metadata_01', N'U') IS NOT NULL
    DROP TABLE dbo.adf_metadata_01
CREATE TABLE dbo.adf_metadata_01
(
    ID INT IDENTITY(1,1) PRIMARY KEY,
	NAME varchar(10),
    update_timestamp DATETIME
);

INSERT INTO dbo.adf_metadata_01
(
	NAME,
	update_timestamp
)
SELECT 'A' AS NAME,CAST('2024-06-01' AS datetime2) AS update_timestamp
UNION ALL
SELECT 'B' AS NAME,CAST('2024-06-02' AS datetime2) AS update_timestamp
UNION ALL
SELECT 'C' AS NAME,CAST('2024-06-03' AS datetime2) AS update_timestamp
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5dc3f81c-67ea-e8df-2f7d-810b0a8c0aea.png)

### 2. メタデータ駆動コピータスクを構築

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f847d8e5-7f55-755d-62f7-b9668471f98c.png)

### 3. メイン制御テーブルの`TopLevelPipelineName`と`TriggerName`の値を変更してテーブル作成を実施

```sql
        "TopLevelPipelineName": "AzureSQLDbToADLSGen2_01",
        "TriggerName": [
            "Trigger_01"
        ],
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d8c3c45d-6a7f-e690-27f7-abe66b1d3537.png)

### 4. `MetadataDrivenCopyTask_xxx_TopLevel`パイプラインをデバック実行しコピーアクティビティが実行されないことを確認

> SELECT count(*) as count FROM adf_metadata.MainControlTable_26j where TopLevelPipelineName='MetadataDrivenCopyTask_26j_TopLevel' and TriggerName like '%Sandbox%' and CopyEnabled = 1

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/88bf43c9-08b8-046a-5c5b-5a86871c94ac.png)

### 5. `MetadataDrivenCopyTask_xxx_TopLevel`パイプラインにて`TopLevelPipelineName`パラメータと`TriggerName`パラメータを追加

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/19572c94-2d1a-f75b-dee7-aacac2911055.png)

### 6. `MetadataDrivenCopyTask_xxx_TopLevel`パイプラインにて`TopLevelPipelineName`変数と`TriggerName`変数を追加

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/76540f77-7001-3df0-850d-1e05e9a48000.png)

### 7. `GetSumOfObjectsToCopy`アクティビティのクエリにて`pipeline().Pipeline`と`pipeline().TriggerName`が利用されている箇所を変数に変更

＜修正前＞

```sql
where TopLevelPipelineName='@{pipeline().Pipeline}' and TriggerName like '%@{pipeline().TriggerName}%'
```

＜修正後＞

```sql
where TopLevelPipelineName='@{variables('TopLevelPipelineName')}' and TriggerName like '%@{variables('TriggerName')}%'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8e9cc2f8-8220-1c16-5f70-802e31af3ded.png)


### 8. `CopyObjectsInOneBatch`アクティビティにてパラメータとして渡している`pipeline().Pipeline`と`pipeline().TriggerName`が利用されている箇所を変数に変更

＜修正前＞

```text
@{pipeline().Pipeline}
```

```text
@{pipeline().TriggerName}
```

＜修正後＞

```text
@variables('TopLevelPipelineName')
```

```text
@variables('TriggerName')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/976a3e2a-280c-5afd-ed76-00ec6d6477e6.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eac0c492-3905-36d9-7b33-c47ccefb62f5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/538b5160-22b4-0a29-8d97-a566ad1fcf2a.png)


### 9. `GetSumOfObjectsToCopy`アクティビティの後に変数の設定を追加

- 名前: 任意
- 変数の種類: パイプライン変数
- 名前： `Set TopLevelPipelineName`
- 値: 下記の式

```text
@if(
    empty(pipeline().parameters.TopLevelPipelineName)
    ,pipeline().Pipeline
    ,pipeline().parameters.TopLevelPipelineName
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fee16c43-ae65-e192-8a01-d6973cfd2ad4.png)

### 10. 再度、`GetSumOfObjectsToCopy`アクティビティの後に変数の設定を追加

- 名前: 任意
- 変数の種類: パイプライン変数
- 名前： `Set TriggerName`
- 値: 下記の式

```text
@if(
    empty(pipeline().parameters.TriggerName)
    ,pipeline().TriggerName
    ,pipeline().parameters.TriggerName
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c1e2f6cc-814b-6812-7b96-114713a1ef0d.png)

### 11. 新しいパイプラインを作成して`Execute Pipeline`アクティビティを追加

- 名前: 任意
- 呼び出されたパイプライン: `MetadataDrivenCopyTask_xxx_TopLevel`のパイプライン
- パラメータに下記を設定
    - TopLevelPipelineName: メイン制御テーブル名
    - TopLevelPipelineName: `AzureSQLDbToADLSGen2_01`
    - TriggerName: `Trigger_01`

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3cb3b365-5b19-4452-3e3b-f52c818ca4d4.png)

### 12. 作成したパイプラインを実行してデータが連携されることを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/62956278-ad85-5157-aaea-d7c98b80e4d7.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fb335622-d54e-ae01-331d-33d6e43c90a1.png)