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

構築手順

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

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
;

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

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

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

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

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

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

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

＜修正前＞

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

＜修正後＞

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

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

＜修正前＞

@{pipeline().Pipeline}

@{pipeline().TriggerName}

＜修正後＞

@variables('TopLevelPipelineName')

@variables('TriggerName')

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

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

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

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

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

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

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

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

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