Azure Data Factory のメタデータ駆動のコピータスクにて`watermarkColumnStartValue`列を単独の列として保持する方法

## 概要

Azure Data Factory のメタデータ駆動のコピータスクにて、`DataLoadingBehaviorSettings`列の json 内に保持されている`watermarkColumnStartValue`を`watermarkColumnStartValue`列を単独の列として保持する方法を共有します。

## 実施手順

### 1. テーブル作成とデータ挿入のスクリプトを修正

テーブル定義にて`watermarkColumnStartValue`列を追加。

```diff_sql
            CREATE TABLE [dbo].[MainControlTable_557](
                [Id] [int] IDENTITY(1,1) NOT NULL PRIMARY KEY,
                [SourceObjectSettings] [nvarchar](max) NULL,
                [SourceConnectionSettingsName] [varchar](max) NULL,
                [CopySourceSettings] [nvarchar](max) NULL,
                [SinkObjectSettings] [nvarchar](max) NULL,
                [SinkConnectionSettingsName] [varchar](max) NULL,
                [CopySinkSettings] [nvarchar](max) NULL,
                [CopyActivitySettings] [nvarchar](max) NULL,
                [TopLevelPipelineName] [varchar](max) NULL,
                [TriggerName] [nvarchar](max) NULL,
                [DataLoadingBehaviorSettings] [nvarchar](max) NULL,
+                watermarkColumnStartValue [nvarchar](max) NULL,
                [TaskId] [int] NULL,
                [CopyEnabled] [bit] NULL
        ) 
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e2cfe413-9482-2c29-765e-921219a766ea.png)


メイン制御テーブルにおける`DataLoadingBehaviorSettings`列の json 内に保持されている`watermarkColumnStartValue`を`watermarkColumnStartValue`列として単独化。

```diff_sql
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "DeltaLoad",
            "watermarkColumnName": "update_timestamp",
            "watermarkColumnType": "DateTime",
-            "watermarkColumnStartValue": "2000-01-01T00:00:00.000Z"
        },
+        "watermarkColumnStartValue": "2000-01-01T00:00:00.000Z",
        "TaskId": 0,
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d849a745-811b-e0f4-eb8a-60f10d6f06d3.png)

### 2. メイン制御テーブルへの更新文を修正

```diff_sql
                UPDATE [dbo].[MainControlTable_557]
-                SET [DataLoadingBehaviorSettings]=JSON_MODIFY([DataLoadingBehaviorSettings],'$.watermarkColumnStartValue', @watermarkColumnStartValue) WHERE Id = @Id
+                SET [watermarkColumnStartValue]=@watermarkColumnStartValue WHERE Id = @Id
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/98301f56-107c-4980-beac-44eafc3843da.png)

### 3. `MetadataDrivenCopyTask_xxx_BottomLevel`における`DeltaLoadOneObject`アクティビティのソースのクエリを変更

```diff_sql
-    json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue, 
+    item().watermarkColumnStartValue, 
-    concat('''', json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue, ''''))}
+    concat('''', item().watermarkColumnStartValue, ''''))}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60e73a89-965c-b94a-4e88-84b08c3626c8.png)

### 3. `MetadataDrivenCopyTask_xxx_BottomLevel`における`DeltaLoadOneObject`アクティビティのソースのクエリを変更

```diff_sql
-    json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue, 
+    item().watermarkColumnStartValue, 
-    concat('''', json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue, ''''))}
+    concat('''', item().watermarkColumnStartValue, ''''))}
```

### 4. `MetadataDrivenCopyTask_xxx_BottomLevel`における`DeltaLoadOneObject`アクティビティのシンク`cw_fileName`のを変更

```diff_sql
- @{json(item().SinkObjectSettings).fileName}-@{json(item().DataLoadingBehaviorSettings).watermarkColumnStartValue}-@{activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue}
+ @{json(item().SinkObjectSettings).fileName}-@{item().watermarkColumnStartValue}-@{activity('GetMaxWatermarkValue').output.firstRow.CurrentMaxWaterMarkColumnValue}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bc0bc3cd-8059-9ead-3685-45aeba83a86b.png)

### 動作確認

#### パイプラインを実行して正常終了することを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6a940e24-400a-866b-06a4-30aa1e175779.png)

### メイン制御テーブルにおける`watermarkColumnStartValue`列が更新されていることを確認

```sql
SELECT * FROM [dbo].[MainControlTable_557]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f26468d9-71b9-e7cb-33d4-bbb196ad0c76.png)