Azure Data Factory のメタデータ駆動のコピータスクにて透かし列（`watermarkColumn`）が Datetime の場合に抽出期間を長くする方法

## 概要

Azure Data Factory のメタデータ駆動のコピータスクにて差分読み込み（`DeltaLoad`）を透かし列（`watermarkColumn`、基準値列ともいう）を Datetime で実施する場合に抽出期間を長くする方法を共有します。実施概要を説明後に検証手順を記述します。

差分読み込みでは、透かし列（例えば`update_timestamp`列）を使用して、前回連携されたデータより新しいデータのみを取得します。このプロセスは、更新日時が秒単位で正確に記録されている場合には問題ありません。しかし、システムによっては日付のみが記録されることがあり、その場合、適切なデータ連携が行えない可能性があります。

```sql
select *
from [dbo].[adf_metadata_01]
where [update_timestamp] > '2024-06-03T00:00:00'
    and [update_timestamp] <= '2024-06-03T00:00:00'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/82e71101-caa7-3a52-1f8a-062aa5d94aec.png)

データ連携期間を延長する目的で、DeltaLoadOneObject アクティビティのクエリ設定を修正し、抽出期間を拡大する方法を確立しました。WaterMarkIncrement パラメータで指定された日数（例：`-1日` ）を加算して、抽出期間を延長します。
```sql
select 
    *
    from [@{json(item().SourceObjectSettings).schema}].[@{json(item().SourceObjectSettings).table}] 
    where [@{json(item().DataLoadingBehaviorSettings).watermarkColumnName}] > @{
        if(
            contains(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'Int'),
        json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
        if(
            equals(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'DateTime'),
            concat(
                'DATEADD(day, ',
                pipeline().parameters.WaterMarkIncrement,
                ' , ',
                '''',
                json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
                '''',
                ')'
            ),
            concat(
                '''',
                json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
                ''''
            )
        )
        )
    }
    and [@{json(item().DataLoadingBehaviorSettings).watermarkColumnName}] <= @{
        if(
            contains(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'Int'),
            activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue,
            concat(
                '''',
                activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue,
                ''''
            )
        )
    }
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/111333b3-bc24-8998-1544-428a7c9f56d9.png)

本手法を適用する主なデメリットは、連携先でソースシステムが保証する主キー制約が失われることです。しかし、コピーアクティビティを通じてデータを連携した後、下記の記事で紹介しているデータアーキテクチャを実装することにより、このデメリットを克服することが可能です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/da0c53d4-033c-6a7f-8541-3fbd87e348d5.png)

引用元：[誰も教えてくれないメダリオンアーキテクチャの デザインメソッド：JEDA データエンジニア分科会 #1 #Python - Qiita](https://qiita.com/manabian/items/57373e833df5b4f65184)

## 検証手順

### 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
;

SELECT
	*
	FROM
		dbo.adf_metadata_01
```

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ca475e56-7475-2f08-5f73-e035286c3321.png)

### 3. `DataLoadingBehaviorSettings`列内の`watermarkColumnStartValue`の値がデフォルトであることを確認

```sql
SELECT 
	ID,
	JSON_VALUE(DataLoadingBehaviorSettings, '$.watermarkColumnStartValue') AS watermarkColumnStartValue
	FROM
		[adf_metadata].[MainControlTable_fj6]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b07645dc-cd78-47f0-e2f8-cab6efb084d3.png)


### 4. Azure Data Factory にてメタデータ駆動のコピータスクを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60a427a0-4b25-adb0-8aa9-4020d1ea32ae.png)

### 5. `DataLoadingBehaviorSettings`列内の`watermarkColumnStartValue`の値が更新されていることを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0dea31c0-8ccc-3b8b-552d-ab570c134b23.png)

### 6. ソースのテーブルにデータを追加

```sql
INSERT INTO dbo.adf_metadata_01
(
	NAME,
	update_timestamp
)
SELECT 'D' AS NAME,CAST('2024-06-03' AS datetime2) AS update_timestamp
;

SELECT
	*
	FROM
		dbo.adf_metadata_01
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/352b6452-d494-5e03-74f3-e250162dbbbb.png)

### 7. Azure Data Factory にてメタデータ駆動のコピータスクを再実行してデータが連携されないことを確認

`書き込まれた行`が`0`となっているため、データ連携されていない。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/07fa4762-d5f3-19d4-7cdf-98d7326a3418.png)

### 8. `MetadataDrivenCopyTask_xxx_ BottomLevel`パイプラインにて`WaterMarkIncrement`パラメータの追加と`DeltaLoadOneObject`アクティビティにてクエリの設定値を変更

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ebb691a1-d5dd-fc95-725f-4c806f50124c.png)

```sql
select 
    *
    from [@{json(item().SourceObjectSettings).schema}].[@{json(item().SourceObjectSettings).table}] 
    where [@{json(item().DataLoadingBehaviorSettings).watermarkColumnName}] > @{
        if(
            contains(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'Int'),
        json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
        if(
            equals(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'DateTime'),
            concat(
                'DATEADD(day, ',
                pipeline().parameters.WaterMarkIncrement,
                ' , ',
                '''',
                json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
                '''',
                ')'
            ),
            concat(
                '''',
                json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue,
                ''''
            )
        )
        )
    }
    and [@{json(item().DataLoadingBehaviorSettings).watermarkColumnName}] <= @{
        if(
            contains(json(item().DataLoadingBehaviorSettings).watermarkColumnType, 'Int'),
            activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue,
            concat(
                '''',
                activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue,
                ''''
            )
        )
    }
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b50fa40-170f-7d8d-6cd6-d139376aac4d.png)

### 9. Azure Data Factory にてメタデータ駆動のコピータスクを再実行してデータが連携されることを確認

`書き込まれた行`が`2`となっているため、`update_timestamp`が`2024-06-03`のレコードである NAME が`C`と`D`の 2 レコードが連携される。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8b42056f-5769-1e9f-c312-3e018070da91.png)