Azure Data Factory のメタデータ駆動のコピータスクにてメタデータを Python （Databricks）環境で生成する方法

## 概要

Azure Data Factory のメタデータ駆動のコピータスクにてメタデータを Python （Databricks）環境で生成する方法を共有します。

メタデータ駆動のコピータスクを利用する場合にはメイン制御テーブルに登録する下記のような json 文字列を作成する必要があるのですが、Azure Data Factory 上で GUI で作成するの際にはとても退屈なときがあります。

```json
[
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_02"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_02",
            "folderPath": "adf_metadata_02",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_0jc_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1
    }
]
```

下記のような Excel から json 形式の文字列を生成すれば開発生産性が高くなると考えました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5cd094be-44b1-6ab9-65fb-9db32861ee17.png)

上記の Excel に記載されている下記のような 3 つのメタデータを生成します。Excel からの生成に加えて、`Id`列を Excel で管理し`watermarkColumnStartValue`列を独立した列とすることで Merge 文により Upsert 処理でメイン制御テーブルを更新できるようにします。本記事では、その手順を紹介します。

<details>
  <summary>クリックして json 形式の文字列を展開</summary>

```json
[
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_01"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_01",
            "folderPath": "adf_metadata_01",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_0jc_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 1,
        "watermarkColumnStartValue": null
    },
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_02"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_02",
            "folderPath": "adf_metadata_02",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_0jc_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 2,
        "watermarkColumnStartValue": null
    },
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_03"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_03",
            "folderPath": "adf_metadata_03",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_0jc_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 3,
        "watermarkColumnStartValue": null
    }
]
```

</details>

## 手順

### 1. Azure Data Factory にてメタデータ駆動コピータスクを作成してメイン制御テーブルに登録する SQL 文をを取得

<details>
  <summary>クリックして メイン制御テーブルに登録する SQL 文を展開</summary>

```sql
            /****** Object:  Table [dbo].[MainControlTable_m8e] ******/
            CREATE TABLE [dbo].[MainControlTable_m8e](
                [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,
                [TaskId] [int] NULL,
                [CopyEnabled] [bit] NULL
        ) 
            DECLARE @MainControlMetadata NVARCHAR(max)  = N'[
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_01"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": null,
            "folderPath": null,
            "fileSystem": "metadata-test"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_m8e_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1
    }
]';
            INSERT INTO [dbo].[MainControlTable_m8e] (
                [SourceObjectSettings],
                [SourceConnectionSettingsName],
                [CopySourceSettings],
                [SinkObjectSettings],
                [SinkConnectionSettingsName],
                [CopySinkSettings],
                [CopyActivitySettings],
                [TopLevelPipelineName],
                [TriggerName],
                [DataLoadingBehaviorSettings],
                [TaskId],
                [CopyEnabled])
            SELECT * FROM OPENJSON(@MainControlMetadata)
                WITH ([SourceObjectSettings] [nvarchar](max) AS JSON,
                [SourceConnectionSettingsName] [varchar](max),
                [CopySourceSettings] [nvarchar](max) AS JSON,
                [SinkObjectSettings] [nvarchar](max) AS JSON,
                [SinkConnectionSettingsName] [varchar](max),
                [CopySinkSettings] [nvarchar](max) AS JSON,
                [CopyActivitySettings] [nvarchar](max) AS JSON,
                [TopLevelPipelineName] [varchar](max),
                [TriggerName] [nvarchar](max) AS JSON,
                [DataLoadingBehaviorSettings] [nvarchar](max) AS JSON,
                [TaskId] [int],
                [CopyEnabled] [bit])
```


</details>

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/23e81c91-4df2-5eab-9961-ef9b6f85a093.png)

### 2. Excel にて json 形式の文字列における修正箇所の一覧を作成

本記事では下記のような Excel ファイルにより置換します。階層になっている項目についてはドット(`.`)により区切り、後述する Python コードにて json 内の文字列を置換します。

- Excel ファイルにて下記を実施
    - `管理テーブル.xlsx`というファイル名とする
    - `管理テーブル`というシート名とする
    - 表を`テーブル1`というテーブルで保持する
- 下記の列を保持
    - Id
    - SourceObjectSettings.schema
    - SourceObjectSettings.table
    - SinkObjectSettings.fileSystem
    - SinkObjectSettings.folderPath
    - SinkObjectSettings.fileName
    - watermarkColumnStartValue


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/82141b34-0eb1-f63b-3897-d21c6070410d.png)

### 3. Databricks Repos を作成して`conf`というディレクトリを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f02d08d0-2cdf-e156-80dc-ff82d021499b.png)

### 4. Excel ファイルをインポート

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f29d4dd9-24e5-2985-357a-4b641db488c5.png)

### 5. `conf`フォルダの 1 つ上の階層に下記のコードをもつファイルをインポート

<details>
  <summary>クリックして Python コードを展開</summary>

```python
# Databricks notebook source
# MAGIC %md
# MAGIC ## データベースオブジェクトを作成

# COMMAND ----------

# MAGIC %md
# MAGIC ## メタデータ駆動コピーアクティビティを追加する方法

# COMMAND ----------

# MAGIC %md
# MAGIC #### ライブラリや変数を定義

# COMMAND ----------

# MAGIC %pip install openpyxl -q
# MAGIC dbutils.library.restartPython()

# COMMAND ----------

import json
import os
import pprint
import pandas as pd
import inspect

from collections import OrderedDict
from openpyxl import load_workbook
from pyspark.sql.functions import expr, max

# COMMAND ----------

# CreateControlTable_xxx.sql における変数の一部を貼り付け
base_json_str_01 = """{
    "SourceObjectSettings": {
        "schema": "dbo",
        "table": "adf_metadata_01"
    },
    "SinkObjectSettings": {
        "compressionCodec": "snappy",
        "fileName": null,
        "folderPath": "test06",
        "fileSystem": "metadata-test"
    },
    "CopySourceSettings": {
        "partitionOption": "None",
        "sqlReaderQuery": null,
        "partitionLowerBound": null,
        "partitionUpperBound": null,
        "partitionColumnName": null,
        "partitionNames": null
    },
    "CopyActivitySettings": {
        "translator": {
            "type": "TabularTranslator",
            "typeConversion": true
        }
    },
    "TopLevelPipelineName": "MetadataDrivenCopyTask_m8e_TopLevel",
    "TriggerName": [
        "Sandbox",
        "Manual"
    ],
    "DataLoadingBehaviorSettings": {
        "dataLoadingBehavior": "FullLoad"
    },
    "TaskId": 0,
    "CopyEnabled": 1
}"""

# base_json_str に追加すべき設定を追加
additional_conf = {
    "Id": 0,
    "watermarkColumnStartValue": "",
}

base_json_dict = json.loads(base_json_str_01)
base_dict = {**base_json_dict, **additional_conf}

sort_json_dict = json.loads(base_json_str_01, object_pairs_hook=OrderedDict)
base_dict = {**sort_json_dict, **additional_conf}
sort_order = list(base_dict.keys())

print(sort_order)
pprint.pprint(base_dict)

# COMMAND ----------

# MAGIC %md
# MAGIC
# MAGIC #### Excel ファイルにより設定値を変更

# COMMAND ----------

# Excel ファイルに関する情報を定義
folder_name = "conf"
file_name = "管理テーブル.xlsx"
sheet_name = "管理テーブル"
table_name = "テーブル1"

# Excel から取得する列の一覧を定義
tgt_columns = [
    "Id",
    "SourceObjectSettings.schema",
    "SourceObjectSettings.table",
    "SinkObjectSettings.fileSystem",
    "SinkObjectSettings.folderPath",
    "SinkObjectSettings.fileName",
    "watermarkColumnStartValue",
]

# ベースとなる設定値にて json 形式の文字列として保持しない列を定義
non_json_cols = [
    "Id",
    "TaskId",
    "CopyEnabled",
    "TopLevelPipelineName",
    "TriggerName",
    "DatabricksNotebook",
    "SwapDataLoadingBehaviorSettings",
    "watermarkColumnStartValue"
]

# folder_name が指定されている場合には / を追加
if folder_name != "":
    folder_name = f"/{folder_name}"

# 現在のディレクトリに基づきファイルパスを定義
current_directory = os.getcwd()
current_dir = current_directory
file_path = f"{current_dir}{folder_name}/{file_name}"

# COMMAND ----------

# Excelファイルからワークブックを読み込む
wb = load_workbook(filename=file_path, data_only=True)

# テーブルのセル範囲を取得する
table = wb[sheet_name].tables["テーブル1"]
data = table.ref

# セル範囲のデータを2次元のリストとして取得する
values = []
for row in wb[sheet_name][data]:
    row_values = [cell.value for cell in row]
    values.append(row_values)

# Pandas データフレームを作成
pdf = pd.DataFrame(values[1:], columns=values[0])
pdf = pdf[tgt_columns]

# Spark データフレームに変換後、カラムの値を調整
df = spark.createDataFrame(pdf)
for col_s in df.schema:
    col_n = col_s.name
    # タイムスタンプ型の場合に、文字列型に変換する
    if col_s.dataType.typeName() == "timestamp":
        df = df.withColumn(
            col_n,
            expr(f"date_format(`{col_n}`, 'yyyy-MM-dd\\'T\\'HH:mm:ss.SSS\\'Z\\'')"),
        )
    # TriggerName 列にて Sandbox を追加して Array 型に変換
    if col_n == "TriggerName":
        df = df.withColumn(col_n, expr(f"array('Sandbox', `{col_n}`)"))

# Spark データフレームを 辞書型変数に変換
config_changes_dict = []
for row in df.collect():
    config_changes_dict.append(row.asDict())

# データフレームの内容を表示
df.display()

# COMMAND ----------

returned_json_dicts = []
for modifed_c in config_changes_dict:
    modified_dict = base_dict.copy()

    # Excel ファイル内容に基づきベースとなる設定値の内容を変更
    for val_k, val_v in modifed_c.items():
        # `.`を含むカラム名の場合、`.`を区切り文字として分割し、辞書型変数にて値を更新する
        tgt_items = val_k.split(".")
        if len(tgt_items) == 1:
            modified_dict[tgt_items[-1]] = val_v
        else:
            value = modified_dict
            for tgt_i in tgt_items[:-1]:
                value = value.setdefault(tgt_i, {})
            value[tgt_items[-1]] = val_v

    # json 形式の文字列として保持すべきカラムにて文字列化
    returned_json_dict = {}
    for col_n in modified_dict.keys():
        if col_n in non_json_cols:
            returned_json_dict[col_n] = modified_dict[col_n]
        else:
            returned_json_dict[col_n] = dict(modified_dict[col_n])
            # Databricks に制御テーブルを保持する場合には下記を有効化
            # returned_json_dict[col_n] = json.dumps(modified_dict[col_n], ensure_ascii=False)

    returned_json_dicts.append(dict(returned_json_dict))

sorted_returned_json_dicts = [{k: d[k] for k in sort_order} for d in returned_json_dicts]
returned_json = json.dumps(
    sorted_returned_json_dicts,
    indent=4,
    ensure_ascii=False,
)
print(returned_json)

# COMMAND ----------

# 書き込み json の情報を定義
folder_name = "conf"
json_file_name = "metadta.json"

# json ファイルとして書き込み
current_directory = os.getcwd()
tgt_directory = os.path.join(current_directory, folder_name)
file_path = os.path.join(tgt_directory, json_file_name)
with open(file_path, 'w') as file:
    file.write(returned_json)

# COMMAND ----------

# end

```

</details>

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/93cf9419-a5af-c2ff-9939-005a3c7352c5.png)

### 6. cmd 6 における`base_json_str_01`変数を取得した単一の json 形式の文字列に変更

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/735d1fc8-a229-316f-a9ad-e8776aeac9fb.png)

### 7. ノートブックを実行し正常終了することを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/12028934-a0d2-dc34-a95d-51f88097637f.png)

### 8. `/conf/metadta.json`ファイルを開き、ファイル内容を記録する

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/87c5379b-b5de-3722-6e44-5387cdd4789f.png)

### 9. テーブル作成文にて`Id`列のIDENTITY プロパティを削除、及び、`watermarkColumnStartValue`列を追加後にテーブルを作成を実行

```diff_sql
CREATE TABLE [dbo].[MainControlTable_m8e](
-    [Id] [int] IDENTITY(1,1) NOT NULL PRIMARY KEY,
+    [Id] [int] 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/a253d408-23f3-568b-e9a9-7fd449310874.png)

### 10. INSERT 文を MERGE 文に変更

```sql
DECLARE @MainControlMetadata NVARCHAR(max)  = N'[]';
MERGE INTO [dbo].[MainControlTable_m8e] AS TARGET
USING (
    SELECT * FROM OPENJSON(@MainControlMetadata)
    WITH (
        [Id] [int],
        [SourceObjectSettings] [nvarchar](max) AS JSON,
        [SourceConnectionSettingsName] [varchar](max),
        [CopySourceSettings] [nvarchar](max) AS JSON,
        [SinkObjectSettings] [nvarchar](max) AS JSON,
        [SinkConnectionSettingsName] [varchar](max),
        [CopySinkSettings] [nvarchar](max) AS JSON,
        [CopyActivitySettings] [nvarchar](max) AS JSON,
        [TopLevelPipelineName] [varchar](max),
        [TriggerName] [nvarchar](max) AS JSON,
        [DataLoadingBehaviorSettings] [nvarchar](max) AS JSON,
        [watermarkColumnStartValue] nvarchar(max),
        [TaskId] [int],
        [CopyEnabled] [bit]
    )
) AS SOURCE
ON TARGET.[Id] = SOURCE.[Id]
WHEN MATCHED THEN
    UPDATE SET
        TARGET.[SourceObjectSettings] = SOURCE.[SourceObjectSettings],
        TARGET.[SourceConnectionSettingsName] = SOURCE.[SourceConnectionSettingsName],
        TARGET.[CopySourceSettings] = SOURCE.[CopySourceSettings],
        TARGET.[SinkObjectSettings] = SOURCE.[SinkObjectSettings],
        TARGET.[SinkConnectionSettingsName] = SOURCE.[SinkConnectionSettingsName],
        TARGET.[CopySinkSettings] = SOURCE.[CopySinkSettings],
        TARGET.[CopyActivitySettings] = SOURCE.[CopyActivitySettings],
        TARGET.[TopLevelPipelineName] = SOURCE.[TopLevelPipelineName],
        TARGET.[TriggerName] = SOURCE.[TriggerName],
        TARGET.[DataLoadingBehaviorSettings] = SOURCE.[DataLoadingBehaviorSettings],
        TARGET.[TaskId] = SOURCE.[TaskId],
        TARGET.[CopyEnabled] = SOURCE.[CopyEnabled]
WHEN NOT MATCHED BY TARGET THEN
    INSERT (
        [Id],
        [SourceObjectSettings],
        [SourceConnectionSettingsName],
        [CopySourceSettings],
        [SinkObjectSettings],
        [SinkConnectionSettingsName],
        [CopySinkSettings],
        [CopyActivitySettings],
        [TopLevelPipelineName],
        [TriggerName],
        [DataLoadingBehaviorSettings],
        [watermarkColumnStartValue],
        [TaskId],
        [CopyEnabled]
    )
    VALUES (
        SOURCE.[Id],
        SOURCE.[SourceObjectSettings],
        SOURCE.[SourceConnectionSettingsName],
        SOURCE.[CopySourceSettings],
        SOURCE.[SinkObjectSettings],
        SOURCE.[SinkConnectionSettingsName],
        SOURCE.[CopySinkSettings],
        SOURCE.[CopyActivitySettings],
        SOURCE.[TopLevelPipelineName],
        SOURCE.[TriggerName],
        SOURCE.[DataLoadingBehaviorSettings],
        SOURCE.[watermarkColumnStartValue],
        SOURCE.[TaskId],
        SOURCE.[CopyEnabled]
    );
```

### 11. `MainControlMetadata`変数に生成した json 形式の文字列をセットした上で実行

<details>
  <summary>クリックして メイン制御テーブルに登録する SQL 文を展開</summary>

```sql
DECLARE @MainControlMetadata NVARCHAR(max)  = N'[
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_01"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_01",
            "folderPath": "adf_metadata_01",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_m8e_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 1,
        "watermarkColumnStartValue": null
    },
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_02"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_02",
            "folderPath": "adf_metadata_02",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_m8e_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 2,
        "watermarkColumnStartValue": null
    },
    {
        "SourceObjectSettings": {
            "schema": "dbo",
            "table": "adf_metadata_03"
        },
        "SinkObjectSettings": {
            "compressionCodec": "snappy",
            "fileName": "adf_metadata_03",
            "folderPath": "adf_metadata_03",
            "fileSystem": "adf-copy"
        },
        "CopySourceSettings": {
            "partitionOption": "None",
            "sqlReaderQuery": null,
            "partitionLowerBound": null,
            "partitionUpperBound": null,
            "partitionColumnName": null,
            "partitionNames": null
        },
        "CopyActivitySettings": {
            "translator": {
                "type": "TabularTranslator",
                "typeConversion": true
            }
        },
        "TopLevelPipelineName": "MetadataDrivenCopyTask_m8e_TopLevel",
        "TriggerName": [
            "Sandbox",
            "Manual"
        ],
        "DataLoadingBehaviorSettings": {
            "dataLoadingBehavior": "FullLoad"
        },
        "TaskId": 0,
        "CopyEnabled": 1,
        "Id": 3,
        "watermarkColumnStartValue": null
    }
]';
MERGE INTO [dbo].[MainControlTable_m8e] AS TARGET
USING (
    SELECT * FROM OPENJSON(@MainControlMetadata)
    WITH (
        [Id] [int],
        [SourceObjectSettings] [nvarchar](max) AS JSON,
        [SourceConnectionSettingsName] [varchar](max),
        [CopySourceSettings] [nvarchar](max) AS JSON,
        [SinkObjectSettings] [nvarchar](max) AS JSON,
        [SinkConnectionSettingsName] [varchar](max),
        [CopySinkSettings] [nvarchar](max) AS JSON,
        [CopyActivitySettings] [nvarchar](max) AS JSON,
        [TopLevelPipelineName] [varchar](max),
        [TriggerName] [nvarchar](max) AS JSON,
        [DataLoadingBehaviorSettings] [nvarchar](max) AS JSON,
        [watermarkColumnStartValue] nvarchar(max),
        [TaskId] [int],
        [CopyEnabled] [bit]
    )
) AS SOURCE
ON TARGET.[Id] = SOURCE.[Id]
WHEN MATCHED THEN
    UPDATE SET
        TARGET.[SourceObjectSettings] = SOURCE.[SourceObjectSettings],
        TARGET.[SourceConnectionSettingsName] = SOURCE.[SourceConnectionSettingsName],
        TARGET.[CopySourceSettings] = SOURCE.[CopySourceSettings],
        TARGET.[SinkObjectSettings] = SOURCE.[SinkObjectSettings],
        TARGET.[SinkConnectionSettingsName] = SOURCE.[SinkConnectionSettingsName],
        TARGET.[CopySinkSettings] = SOURCE.[CopySinkSettings],
        TARGET.[CopyActivitySettings] = SOURCE.[CopyActivitySettings],
        TARGET.[TopLevelPipelineName] = SOURCE.[TopLevelPipelineName],
        TARGET.[TriggerName] = SOURCE.[TriggerName],
        TARGET.[DataLoadingBehaviorSettings] = SOURCE.[DataLoadingBehaviorSettings],
        TARGET.[TaskId] = SOURCE.[TaskId],
        TARGET.[CopyEnabled] = SOURCE.[CopyEnabled]
WHEN NOT MATCHED BY TARGET THEN
    INSERT (
        [Id],
        [SourceObjectSettings],
        [SourceConnectionSettingsName],
        [CopySourceSettings],
        [SinkObjectSettings],
        [SinkConnectionSettingsName],
        [CopySinkSettings],
        [CopyActivitySettings],
        [TopLevelPipelineName],
        [TriggerName],
        [DataLoadingBehaviorSettings],
        [watermarkColumnStartValue],
        [TaskId],
        [CopyEnabled]
    )
    VALUES (
        SOURCE.[Id],
        SOURCE.[SourceObjectSettings],
        SOURCE.[SourceConnectionSettingsName],
        SOURCE.[CopySourceSettings],
        SOURCE.[SinkObjectSettings],
        SOURCE.[SinkConnectionSettingsName],
        SOURCE.[CopySinkSettings],
        SOURCE.[CopyActivitySettings],
        SOURCE.[TopLevelPipelineName],
        SOURCE.[TriggerName],
        SOURCE.[DataLoadingBehaviorSettings],
        SOURCE.[watermarkColumnStartValue],
        SOURCE.[TaskId],
        SOURCE.[CopyEnabled]
    );
```

</details>

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/09013aa7-0278-857d-9bf1-82e64adff04e.png)

### 12. メイン制御テーブルにデータが登録されたことを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/13b2eaec-980c-a2dd-d869-e29f44e28806.png)

## 動作確認

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/53a49be3-026e-275e-2907-1847cbd7c3a7.png)


### 2. データが書きこまれたことを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2f962ca2-1bb0-e6bf-8702-791d57049713.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2c9c26b3-49ef-b88a-a6f7-89b2bbb2dd86.png)