Azure Data Factory のメタデータ駆動のコピータスクにてデプロイメント環境に応じて接続情報を切り替える方法

概要

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レコード連携されることを確認