概要

Azure Data Factory のメタデータ駆動のコピータスクにてデプロイメント環境（開発や本番などの環境）に応じて接続情報を切り替える方法を共有します。メタデータ駆動のコピータスクのデフォルトで作成される接続制御テーブル（ConnectionControlTable）に、デプロイメント環境（DeployEnv）列を追加する手順を紹介します。

メタデータ駆動を作成する際にリンクサービスに対してパラメータを追加すること、接続制御テーブルの DDL と INSERT 文が生成されます。

パラメーター化されリンクされたサービスを使用した場合、接続制御テーブルはデータストアの接続値を格納します。

引用元：データのコピーツールでメタデータ駆動型の方法を使用して大規模なデータコピーパイプラインを作成する - Azure Data Factory | Microsoft Learn

接続制御テーブルには下記の２つのカラムと連番が付与された Id 列で構成されており、メタデータ駆動のコピータスクのパイプラインにおけるGetSourceConnectionValuesアクティビティにてName列によりレコードを取得しす。接続制御テーブルにデプロイメント環境（DeployEnv）列を追加、および、GetSourceConnectionValuesアクティビティにおける WHERE 句への条件追加を実施することにより、デプロイメント環境に応じて接続情報を切り替える方法を紹介します。

引用元：Build large-scale data copy pipelines with metadata-driven approach in copy data tool - Azure Data Factory | Microsoft Learn

手順

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

１つ目の Azure SQL Dataabse （dev環境相当）にて下記の 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
;

SELECT
	*
	FROM
		dbo.adf_metadata_01

２つ目の Azure SQL Dataabse （prd環境相当）に下記の 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
;

SELECT
	*
	FROM
		dbo.adf_metadata_01

2. パラメータ化したリンクドサービスを新規で作成する手順を含むメタデータ駆動のコピージョブを作成

パラメータとしてserer_nameとdb_nameを作成し、完全修飾ドメイン名とデータベース名にそのパラメータを設定します。

3. 接続管理テーブルのコードをデプロイメント環境（`DeployEnv`）列に関する修正後に実行

/****** Object:  Table [adf_metadata].[ConnectionControlTable_fuy] ******/
CREATE TABLE [adf_metadata].[ConnectionControlTable_fuy](
    [Id] [int] IDENTITY(1,1) NOT NULL PRIMARY KEY,
    [Name] [varchar](max) NOT NULL,
    [DeployEnv] [nvarchar](10) NULL,
    [ConnectionSettings] [nvarchar](max) NULL
) 

DECLARE @ConnectionMetadata NVARCHAR(max)  = N'[
    {
        "Name": "AzureSqlDatabase3_Source_fuy",
        "DeployEnv": "dev",
        "ConnectionSettings": {
            "serer_name": "aqdb-dev.database.windows.net",
            "db_name": "adf-metadata-dev"
        }
    },
    {
        "Name": "AzureSqlDatabase3_Source_fuy",
        "DeployEnv": "prd",
        "ConnectionSettings": {
            "serer_name": "aqdb-pr.database.windows.net",
            "db_name": "adf-metadata-prd"
        }
    }
]';
INSERT INTO [adf_metadata].[ConnectionControlTable_fuy] (
    [Name],
	[DeployEnv],
    [ConnectionSettings])
SELECT * FROM OPENJSON(@ConnectionMetadata)
    WITH (
		[Name] [varchar](max),
		[DeployEnv] [varchar](10),
        [ConnectionSettings] [nvarchar](max) AS JSON
	)
;

4. メイン制御テーブルを作成

5. メタデータ駆動のコピータスクにより作成されたデータセットにてリンクサービスにパラメータを渡すように修正

6. メタデータ駆動のコピータスクの３つのパイプラインにて`DeployEnv`というパラメータを追加

7. `MetadataDrivenCopyTask_xxx_TopLevel`パイプラインと`MetadataDrivenCopyTask_xxx_MiddleLevel`パイプラインにて Execute Pipeline アクティビティにて`DeployEnv`パラメータを渡すように修正

8. `MetadataDrivenCopyTask_xxx_BottomLevel`パイプラインの`GetSourceConnectionValues`アクティビティのクエリの Where 句に`DeployEnv`列のロジックを追加

AND DeployEnv = '@{pipeline().parameters.DeployEnv}'

動作検証

1. `DeployEnv`を`dev`でデバック実行して1レコード連携されることを確認

2. `DeployEnv`を`prd`でデバック実行して3レコード連携されることを確認