概要

Azure Data Factory のメタデータ駆動のコピータスクにて差分読み込み（DeltaLoad）時にCopySourceSettings列のsqlReaderQueryの値を利用する方法を共有します。

全体の読み込み（FullLoad）の際には SQL を記述することでき、その値はCopySourceSettings列のsqlReaderQueryに値がセットされます。この値は差分読み込み（DeltaLoad）の場合には値がセットされません。

差分読み込みにおいても同様に SQL を記述できることで利便性が向上するため、その実装方法を記述します。本手順のデメリットとしては、CopySourceSettings列のsqlReaderQueryの値をセットする必要があることです。

実施手順

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
;

SELECT
	*
	FROM
		dbo.adf_metadata_01

2. Azure Data Factory にて差分読み込みのメタデータ駆動のコピータスクを作成

3. メイン制御テーブルに登録する SQL にて `CopySourceSettings`列の`sqlReaderQuery`の値を変更後にテーブルとして登録

<修正前>

        "CopySourceSettings": {
            "isolationLevel": "ReadUncommitted",
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },

<修正後>

        "CopySourceSettings": {
            "isolationLevel": "ReadUncommitted",
            "partitionOption": "None",
            "sqlReaderQuery": "SELECT ID,NAME FROM [dbo].[adf_metadata_01]",
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },

4. `MetadataDrivenCopyTask_xxx_BottomLevel`パイプラインにおける`DeltaLoadOneObjectアクティビティ`のクエリの一部を修正

<修正前>

select * from [@{json(item().SourceObjectSettings).schema}].[@{json(item().SourceObjectSettings).table}]

<修正後>

@{json(item().CopySourceSettings).sqlReaderQuery}

5. Azure Data Factory にてメタデータ駆動のコピータスクを実行後に`DeltaLoadOneObject`アクティビティの`sqlReaderQuery`の値が想定通りであることを確認

SELECT ID,NAME FROM [dbo].[adf_metadata_01]
    where [update_timestamp] > '2024-06-03T00:00:00'
    and [update_timestamp] <= '2024-06-03T00:00:00