Azure Data Factory または Azure Synapse Analytics を使用して Azure SQL Database のデータをコピーおよび変換する

適用対象: Azure Data Factory Azure Synapse Analytics

この記事では、Azure Data Factory または Azure Synapse パイプラインで Copy アクティビティを使用して、Azure SQL Database との間でデータをコピーし、Data Flow を使用して Azure SQL Database のデータを変換する方法について説明します。 詳細については、Azure Data Factory または Azure Synapse Analytics の概要記事を参照してください。

サポートされる機能

この Azure SQL Database コネクタは、次の機能でサポートされます。

① Azure 統合ランタイム ② セルフホステッド統合ランタイム

コピー アクティビティについて、この Azure SQL Database コネクタは以下の機能をサポートします。

• SQL 認証および Microsoft Entra アプリケーション トークン認証と、Azure リソースのサービス プリンシパルまたはマネージド ID を使って、データをコピーする。
• ソースとして、SQL クエリまたはストアド プロシージャを使用してデータを取得する。 Azure SQL Database ソースからの並列コピーを選択することもできます。詳細については、「SQL Database からの並列コピー」セクションを参照してください。
• シンクとして、ソース スキーマに基づいて、宛先テーブルが存在しない場合はこれを自動的に作成する。テーブルにデータを追加するか、コピー中にカスタム ロジックを使用してストアド プロシージャを呼び出す。

Azure SQL Database のサーバーレス レベルを使用している場合、サーバーが一時停止すると、自動再開の準備が整うまで待機せず、アクティビティの実行が失敗することに注意してください。 実際の実行時に確実にサーバーを稼働した状態にするには、アクティビティの再試行を追加するか、追加のアクティビティを連鎖させます。

はじめに

パイプラインでコピー アクティビティを実行するには、次のいずれかのツールまたは SDK を使用します。

• データ コピー ツール
• Azure ポータル
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure Resource Manager テンプレート

UI を使用して Azure SQL Database のリンク サービスを作成する

次の手順を使用して、Azure portal UI で Azure SQL Database のリンク サービスを作成します。

1. Azure Data Factory または Synapse ワークスペースの [管理] タブに移動し、[リンクされたサービス] を選択して、[新規] をクリックします。

   • Azure Data Factory
   • Azure Synapse

   

2. SQL を検索し、Azure SQL Database コネクタを選択します。

   

3. サービスの詳細を構成し、接続をテストして、新しいリンク サービスを作成します。

   

コネクタの構成の詳細

以下のセクションでは、Azure SQL Database コネクタに固有の Azure Data Factory または Synapse パイプライン エンティティの定義に使用されるプロパティについて詳しく説明します。

リンクされたサービス プロパティ

Azure SQL Database コネクタの推奨バージョンでは、TLS 1.3 がサポートされています。 Azure SQL Database コネクタのバージョンをレガシのものからアップグレードするには、このセクションを参照してください。 プロパティの詳細については、対応するセクションを参照してください。

• 推奨されるバージョン
• レガシ バージョン

推奨されるバージョン

推奨バージョンを適用すると、Azure SQL Database のリンク サービスで次の汎用プロパティがサポートされます。

その他の接続プロパティについては、次の表を参照してください。

SQL 認証

SQL 認証を使用するには、前のセクションで説明した汎用プロパティに加えて、次のプロパティを指定します。

例: SQL 認証の使用

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

例: Azure Key Vault 内のパスワード

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

例: Always Encrypted の使用

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            },
            "alwaysEncryptedSettings": {
                "alwaysEncryptedAkvAuthType": "ServicePrincipal",
                "servicePrincipalId": "<service principal id>",
                "servicePrincipalKey": {
                    "type": "SecureString",
                    "value": "<service principal key>"
                }
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

サービス プリンシパルの認証

前のセクションで説明した汎用的なプロパティに加えて、サービス プリンシパルの認証を使用するには、次のプロパティを指定します。

以下の手順に従う必要もあります。

1. Azure portal から Microsoft Entra アプリケーションを作成します。 アプリケーション名と、リンクされたサービスを定義する次の値を記録しておきます。

   • アプリケーション ID
   • アプリケーション キー
   • テナント ID

2. まだ行っていない場合は、Azure portal でサーバーの Microsoft Entra 管理者をプロビジョニングします。 Microsoft Entra 管理者には Microsoft Entra ユーザーまたは Microsoft Entra グループを指定できますが、サービス プリンシパルは指定できません。 このステップは、次のステップで Microsoft Entra ID を使ってサービス プリンシパルの包含データベース ユーザーを作成できるようにするために行われます。

3. サービス プリンシパルの包含データベース ユーザーを作成します。 SQL Server Management Studio のようなツールと、少なくとも ALTER ANY USER アクセス許可を持つ Microsoft Entra ID を使用して、データをコピーするデータベースに接続します。 次の T-SQL を実行します。

   CREATE USER [your application name] FROM EXTERNAL PROVIDER;
   

4. SQL ユーザーや他のユーザーに対する通常の方法で、サービス プリンシパルに必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。

   ALTER ROLE [role name] ADD MEMBER [your application name];
   

5. Azure Data Factory または Synapse ワークスペースで、Azure SQL Database のリンク サービスを構成します。

サービス プリンシパル認証を使うリンクされたサービスの例

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "hostNameInCertificate": "<host name>",
            "authenticationType": "ServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredential": {
                "type": "SecureString",
                "value": "<application key>"
            },
            "tenant": "<tenant info, e.g. microsoft.onmicrosoft.com>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

システム割り当てマネージド ID 認証

データ ファクトリまたは Synapse ワークスペースは、Azure の他のリソースに対して認証するときにそのサービスを表す、Azure リソースのシステム割り当てマネージド ID に関連付けることができます。 このマネージド ID を Azure SQL Database の認証に使用できます。 指定されたファクトリまたは Synapse ワークスペースでは、この ID を使用してデータベースにアクセスし、データベースとの間でデータをコピーできます。

システム割り当てマネージド ID 認証を使用するには、前のセクションで説明した汎用プロパティを指定し、次の手順に従います。

1. まだ行っていない場合は、Azure portal でサーバーの Microsoft Entra 管理者をプロビジョニングします。 Microsoft Entra ユーザーまたは Microsoft Entra グループを Microsoft Entra 管理者にすることができます。 マネージド ID を持つグループに管理者ロールを付与する場合は、ステップ 3 と 4 をスキップします。 管理者はデータベースに対してフル アクセス権を持っています。

2. マネージド ID の包含データベース ユーザーを作成します。 SQL Server Management Studio のようなツールと、少なくとも ALTER ANY USER アクセス許可を持つ Microsoft Entra ID を使用して、データをコピーするデータベースに接続します。 次の T-SQL を実行します。

   CREATE USER [your_resource_name] FROM EXTERNAL PROVIDER;
   

3. 通常の SQL ユーザーなどと同様に、マネージド ID に必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。

   ALTER ROLE [role name] ADD MEMBER [your_resource_name];
   

4. Azure SQL Database のリンク サービスを構成します。

例

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SystemAssignedManagedIdentity"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

ユーザー割り当てマネージド ID 認証

データ ファクトリまたは Synapse ワークスペースは、Azure の他のリソースに対して認証するときにそのサービスを表す、ユーザー割り当てマネージド ID に関連付けることができます。 このマネージド ID を Azure SQL Database の認証に使用できます。 指定されたファクトリまたは Synapse ワークスペースでは、この ID を使用してデータベースにアクセスし、データベースとの間でデータをコピーできます。

前のセクションで説明した汎用的なプロパティに加えて、ユーザー割り当てマネージド ID 認証を使用するには、次のプロパティを指定します。

以下の手順に従う必要もあります。

1. まだ行っていない場合は、Azure portal でサーバーの Microsoft Entra 管理者をプロビジョニングします。 Microsoft Entra ユーザーまたは Microsoft Entra グループを Microsoft Entra 管理者にすることができます。 ユーザー割り当てマネージド ID を持つグループに管理者ロールを付与する場合は、ステップ 3 をスキップします。 管理者はデータベースに対してフル アクセス権を持っています。

2. ユーザー割り当てマネージド ID の包含データベース ユーザーを作成します。 SQL Server Management Studio のようなツールと、少なくとも ALTER ANY USER アクセス許可を持つ Microsoft Entra ID を使用して、データをコピーするデータベースに接続します。 次の T-SQL を実行します。

   CREATE USER [your_resource_name] FROM EXTERNAL PROVIDER;
   

3. 1 つ以上のユーザー割り当てマネージド ID を作成し、通常の SQL ユーザーなどと同様に、ユーザー割り当てマネージド ID に必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。

   ALTER ROLE [role name] ADD MEMBER [your_resource_name];
   

4. 1 つ以上のユーザー割り当てマネージド ID をデータ ファクトリに割り当てて、ユーザー割り当てマネージド ID ごとに資格情報を作成します。

5. Azure SQL Database のリンク サービスを構成します。

例

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "UserAssignedManagedIdentity",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

レガシ バージョン

レガシ バージョンを適用すると、Azure SQL Database のリンク サービスで次の汎用プロパティがサポートされます。

さまざまな認証の種類に対する固有のプロパティと前提条件については、それぞれ以降のセクションをご覧ください。

• レガシ バージョンの SQL 認証
• レガシ バージョンのサービス プリンシパル認証
• レガシ バージョンのシステム割り当てマネージド ID 認証
• レガシ バージョンのユーザー割り当てマネージド ID 認証

レガシ バージョンの SQL 認証

SQL 認証を使用するには、前のセクションで説明した汎用プロパティを指定します。

レガシ バージョンのサービス プリンシパル認証

前のセクションで説明した汎用的なプロパティに加えて、サービス プリンシパルの認証を使用するには、次のプロパティを指定します。

また、サービス プリンシパル認証の手順に従って、対応するアクセス許可を付与する必要もあります。

レガシ バージョンのシステム割り当てマネージド ID 認証

システム割り当てマネージド ID 認証を使用するには、「システム割り当てマネージド ID 認証」の推奨バージョンと同じ手順に従います。

レガシ バージョンのユーザー割り当てマネージド ID 認証

ユーザー割り当てマネージド ID 認証を使用するには、「ユーザー割り当てマネージド ID 認証」の推奨バージョンと同じ手順に従います。

データセットのプロパティ

データセットの定義に使用できるセクションとプロパティの詳細な一覧については、データセットに関するページを参照してください。

Azure SQL Database データセットでは、次のプロパティがサポートされます。

データセットのプロパティの例

{
    "name": "AzureSQLDbDataset",
    "properties":
    {
        "type": "AzureSqlTable",
        "linkedServiceName": {
            "referenceName": "<Azure SQL Database linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [ < physical schema, optional, retrievable during authoring > ],
        "typeProperties": {
            "schema": "<schema_name>",
            "table": "<table_name>"
        }
    }
}

コピー アクティビティのプロパティ

アクティビティの定義に利用できるセクションとプロパティの完全な一覧については、パイプラインに関するページを参照してください。 このセクションでは、Azure SQL Database のソースとシンクでサポートされるプロパティの一覧を示します。

ソースとしての Azure SQL Database

Azure SQL Database からデータをコピーするために、コピー アクティビティの source セクションでは次のプロパティがサポートされています。

以下の点に注意してください。

• AzureSqlSource に sqlReaderQuery が指定されている場合、コピー アクティビティでは、データを取得するために Azure SQL Database ソースに対してこのクエリを実行します。 sqlReaderStoredProcedureName と storedProcedureParameters を指定して、ストアド プロシージャを指定することもできます (ストアド プロシージャでパラメーターを使用する場合)。
• ソースのストアド プロシージャを使用してデータを取得する場合、異なるパラメーター値が渡されたときにストアド プロシージャが別のスキーマを返すように設計されていると、UI からスキーマをインポートするとき、または自動テーブル作成を使用して SQL データベースにデータをコピーするときに、エラーが発生したり、予期しない結果が表示されます。

SQL クエリの例

"activities":[
    {
        "name": "CopyFromAzureSQLDatabase",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Azure SQL Database input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureSqlSource",
                "sqlReaderQuery": "SELECT * FROM MyTable"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

ストアド プロシージャの例

"activities":[
    {
        "name": "CopyFromAzureSQLDatabase",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Azure SQL Database input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureSqlSource",
                "sqlReaderStoredProcedureName": "CopyTestSrcStoredProcedureWithParameters",
                "storedProcedureParameters": {
                    "stringData": { "value": "str3" },
                    "identifier": { "value": "$$Text.Format('{0:yyyy}', <datetime parameter>)", "type": "Int"}
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

ストアド プロシージャの定義

CREATE PROCEDURE CopyTestSrcStoredProcedureWithParameters
(
    @stringData varchar(20),
    @identifier int
)
AS
SET NOCOUNT ON;
BEGIN
     select *
     from dbo.UnitTestSrcTable
     where dbo.UnitTestSrcTable.stringData != stringData
    and dbo.UnitTestSrcTable.identifier != identifier
END
GO

シンクとしての Azure SQL Database

データを Azure SQL Database にコピーするために、コピー アクティビティの sink セクションでは次のプロパティがサポートされています。

例 1: データを追加する

"activities":[
    {
        "name": "CopyToAzureSQLDatabase",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure SQL Database output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureSqlSink",
                "tableOption": "autoCreate",
                "writeBatchSize": 100000
            }
        }
    }
]

例 2: コピー中にストアド プロシージャを呼び出す

詳しくは、「SQL シンクからのストアド プロシージャの呼び出し」をご覧ください。

"activities":[
    {
        "name": "CopyToAzureSQLDatabase",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure SQL Database output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureSqlSink",
                "sqlWriterStoredProcedureName": "CopyTestStoredProcedureWithParameters",
                "storedProcedureTableTypeParameterName": "MyTable",
                "sqlWriterTableType": "MyTableType",
                "storedProcedureParameters": {
                    "identifier": { "value": "1", "type": "Int" },
                    "stringData": { "value": "str1" }
                }
            }
        }
    }
]

例 3: データをアップサートする

"activities":[
    {
        "name": "CopyToAzureSQLDatabase",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure SQL Database output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureSqlSink",
                "tableOption": "autoCreate",
                "writeBehavior": "upsert",
                "upsertSettings": {
                    "useTempDB": true,
                    "keys": [
                        "<column name>"
                    ]
                },
            }
        }
    }
]

SQL データベースからの並列コピー

Azure SQL Database コネクタでは、コピー アクティビティの際に、データを並列でコピーするための組み込みのデータ パーティション分割が提供されます。 データ パーティション分割オプションは、コピー アクティビティの [ソース] タブにあります。

パーティション分割されるコピーを有効にすると、コピー アクティビティによって Azure SQL Database ソースに対する並列クエリが実行され、パーティションごとにデータが読み込まれます。 並列度は、コピー アクティビティの parallelCopies 設定によって制御されます。 たとえば、parallelCopies を 4 に設定した場合、指定したパーティション オプションと設定に基づいて 4 つのクエリが同時に生成され、実行されます。各クエリでは、Azure SQL Database からデータの一部を取得します。

Azure SQL Database から大量のデータを読み込む場合は特に、データ パーティション分割を行う並列コピーを有効にすることが推奨されます。 さまざまなシナリオの推奨構成を以下に示します。 ファイルベースのデータ ストアにデータをコピーする場合は、複数のファイルとしてフォルダーに書き込む (フォルダー名のみを指定する) ことをお勧めします。この場合、1 つのファイルに書き込むよりもパフォーマンスが優れています。

パーティション オプションを使用してデータを読み込む場合のベスト プラクティス:

1. データ スキューを回避するため、パーティション列 (主キーや一意キーなど) には特徴のある列を選択します。
2. テーブルに組み込みパーティションがある場合は、パフォーマンスを向上させるためにパーティション オプションとして "テーブルの物理パーティション" を使用します。
3. Azure Integration Runtime を使用してデータをコピーする場合は、より大きな (4 より大きい) "データ統合単位 (DIU)" (>4) を設定すると、より多くのコンピューティング リソースを利用できます。 そこで、該当するシナリオを確認してください。
4. パーティション数は、"コピーの並列処理の次数" によって制御されます。この数値を大きくしすぎるとパフォーマンスが低下するため、この数値は、(DIU またはセルフホステッド IR ノードの数) x (2 から 4) に設定することをお勧めします。

例: 複数の物理パーティションがある大きなテーブル全体から読み込む

"source": {
    "type": "AzureSqlSource",
    "partitionOption": "PhysicalPartitionsOfTable"
}

例: 動的範囲パーティションを使用してクエリを実行する

"source": {
    "type": "AzureSqlSource",
    "query": "SELECT * FROM <TableName> WHERE ?DfDynamicRangePartitionCondition AND <your_additional_where_clause>",
    "partitionOption": "DynamicRange",
    "partitionSettings": {
        "partitionColumnName": "<partition_column_name>",
        "partitionUpperBound": "<upper_value_of_partition_column (optional) to decide the partition stride, not as data filter>",
        "partitionLowerBound": "<lower_value_of_partition_column (optional) to decide the partition stride, not as data filter>"
    }
}

物理パーティションを確認するためのサンプル クエリ

SELECT DISTINCT s.name AS SchemaName, t.name AS TableName, pf.name AS PartitionFunctionName, c.name AS ColumnName, iif(pf.name is null, 'no', 'yes') AS HasPartition
FROM sys.tables AS t
LEFT JOIN sys.objects AS o ON t.object_id = o.object_id
LEFT JOIN sys.schemas AS s ON o.schema_id = s.schema_id
LEFT JOIN sys.indexes AS i ON t.object_id = i.object_id 
LEFT JOIN sys.index_columns AS ic ON ic.partition_ordinal > 0 AND ic.index_id = i.index_id AND ic.object_id = t.object_id 
LEFT JOIN sys.columns AS c ON c.object_id = ic.object_id AND c.column_id = ic.column_id 
LEFT JOIN sys.partition_schemes ps ON i.data_space_id = ps.data_space_id 
LEFT JOIN sys.partition_functions pf ON pf.function_id = ps.function_id 
WHERE s.name='[your schema]' AND t.name = '[your table name]'

テーブルに物理パーティションがある場合、次のように、"HasPartition" は "yes" と表示されます。

Azure SQL Database にデータを読み込む際のベスト プラクティス

Azure SQL Database にデータをコピーする場合は、さまざまな書き込み動作が必要になることがあります。

• 追加: ソース データには新しいレコードのみが含まれている。
• アップサート: ソース データには挿入と更新の両方が含まれている。
• 上書き: 毎回ディメンション テーブル全体を再度読み込みたい。
• カスタム ロジックでの書き込み: 宛先テーブルへの最終挿入の前に追加の処理が必要である。

サービスで構成する方法とベスト プラクティスについては、対応するセクションを参照してください。

データを追加する

データの追加は、この Azure SQL Database シンク コネクタの既定の動作です。 サービスでは、テーブルに効率的に書き込むために一括挿入が実行されます。 コピー アクティビティで、それに応じてソースとシンクを構成できます。

データをアップサートする

Copy アクティビティは、データベースの一時テーブルにデータを読み込み、キーが存在する場合はシンク テーブルのデータを更新し、そうでない場合は新しいデータを挿入する処理をネイティブにサポートするようになりました。 コピー アクティビティでのアップサート設定の詳細については、シンクとしての Azure SQL Databaseに関するページを参照してください。

テーブル全体を上書きする

コピー アクティビティ シンクで preCopyScript プロパティを構成できます。 この場合、実行される Copy アクティビティごとに、サービスで最初にスクリプトが実行されます。 次に、コピーが実行されてデータが挿入されます。 たとえば、テーブル全体を最新のデータで上書きするには、ソースから新しいデータを一括で読み込む前に、すべてのレコードを最初に削除するスクリプトを指定します。

カスタム ロジックでデータを書き込む

カスタム ロジックでデータを書き込む手順は、「データをアップサートする」セクションで説明されている手順に似ています。 宛先テーブルへのソース データの最終挿入の前に追加の処理を適用する必要がある場合は、ステージング テーブルに読み込んだ後、ストアド プロシージャ アクティビティを呼び出すか、コピー アクティビティのシンクのストアド プロシージャを呼び出してデータを適用するか、マッピング データ フローを使用できます。

SQL シンクからのストアド プロシージャの呼び出し

Azure SQL Database にデータをコピーするときに、ユーザーが指定したストアド プロシージャを構成し、ソース テーブルの各バッチに関する追加のパラメーターと共に呼び出すこともできます。 ストアド プロシージャ機能は テーブル値パラメーターを利用しています。

組み込みのコピー メカニズムでは目的を達成できない場合は、ストアド プロシージャを使用できます。 1 つの例は、宛先テーブルへのソース データの最終挿入の前に追加の処理を適用する場合です。 その他の処理の例をいくつか挙げると、列のマージ、追加の値の検索、複数のテーブルへの挿入があります。

次の例では、Azure SQL Database 内のテーブルに upsert を行うストアド プロシージャを使用する方法を示します。 入力データとシンクの Marketing テーブルには、それぞれ 3 つの列 (ProfileID、State、Category) があるものとします。 ProfileID 列に基づいてアップサートを行い、"ProductA" という特定のカテゴリに対してのみ適用します。

1. データベースで、sqlWriterTableType と同じ名前のテーブル型を定義します。 テーブル型のスキーマは、入力データから返されるスキーマと同じです。

   CREATE TYPE [dbo].[MarketingType] AS TABLE(
       [ProfileID] [varchar](256) NOT NULL,
       [State] [varchar](256) NOT NULL,
       [Category] [varchar](256) NOT NULL
   )
   

2. データベース内で、sqlWriterStoredProcedureName と同じ名前のストアド プロシージャを定義します。 これによって指定したソースの入力データが処理され、出力テーブルにマージされます。 ストアド プロシージャ内のテーブル型のパラメーター名は、データセットで定義されている tableName と同じです。

   CREATE PROCEDURE spOverwriteMarketing @Marketing [dbo].[MarketingType] READONLY, @category varchar(256)
   AS
   BEGIN
   MERGE [dbo].[Marketing] AS target
   USING @Marketing AS source
   ON (target.ProfileID = source.ProfileID and target.Category = @category)
   WHEN MATCHED THEN
       UPDATE SET State = source.State
   WHEN NOT MATCHED THEN
       INSERT (ProfileID, State, Category)
       VALUES (source.ProfileID, source.State, source.Category);
   END
   

3. Azure Data Factory または Synapse パイプラインで、Copy アクティビティの SQL シンク セクションを次のように定義します。

   "sink": {
       "type": "AzureSqlSink",
       "sqlWriterStoredProcedureName": "spOverwriteMarketing",
       "storedProcedureTableTypeParameterName": "Marketing",
       "sqlWriterTableType": "MarketingType",
       "storedProcedureParameters": {
           "category": {
               "value": "ProductA"
           }
       }
   }
   

ストアド プロシージャを使用して Azure SQL Database にデータを書き込む場合は、シンクによってソース データがミニ バッチに分割され、挿入が実行されるため、ストアド プロシージャ内の余分なクエリを複数回実行できます。 Azure SQL Database にデータを書き込む前に実行するコピー アクティビティのクエリがある場合は、ストアド プロシージャに追加することはお勧めしません。[コピー前スクリプト] ボックスに追加します。

Mapping Data Flow のプロパティ

マッピング データ フローでデータを変換する場合、Azure SQL Database からテーブルの読み取りと書き込みを実行できます。 詳細については、マッピング データ フローのソース変換とシンク変換に関する記事をご覧ください。

ソース変換

Azure SQL Database に固有の設定は、ソース変換の [Source Options](ソース オプション) タブにあります。

[Input]: テーブルにあるソースを指す (Select * from <table-name> に相当) かカスタム SQL クエリを入力するかを選択します。

Query: [Input] フィールドで [Query] を選択した場合は、ソースに対する SQL クエリを入力します。 この設定により、データセットで選択したすべてのテーブルがオーバーライドされます。 ここでは Order By 句はサポートされていませんが、完全な SELECT FROM ステートメントを設定することができます。 ユーザー定義のテーブル関数を使用することもできます。 select * from udfGetData() は、テーブルを返す SQL の UDF です。 このクエリでは、お使いのデータ フローで使用できるソース テーブルが生成されます。 テスト対象またはルックアップ対象の行を減らすうえでも、クエリの使用は有効な手段です。

CREATE PROC CTESP @query nvarchar(max)
AS
BEGIN
EXECUTE sp_executesql @query;
END

[Stored procedure] : ソース データベースから実行されるストアド プロシージャからプロジェクションおよびソース データを生成する場合は、このオプションを選択します。 スキーマ、プロシージャ名、パラメーターを入力することも、[Refresh](最新の情報に更新) をクリックして、スキーマとプロシージャ名の検出をサービスに要求することもできます。 次に、[Import ] をクリックして、すべてのプロシージャ パラメーターをフォーム @paraName を使用してインポートできます。

• SQL の例: Select * from MyTable where customerId > 1000 and customerId < 2000
• パラメーター化された SQL の例: "select * from {$tablename} where orderyear > {$year}"

Batch size: 大量データを読み取りにまとめるバッチ サイズを入力します。

Isolation Level: マッピング データ フローでの SQL ソースの既定値は [コミットされていないものを読み取り] です。 ここで分離レベルを次のいずれかの値に変更できます。

• コミットされたものを読み取り
• コミットされていないものを読み取り
• 反復可能読み取り
• シリアル化可能
• なし (分離レベルを無視)

増分抽出を有効にする: このオプションを使用して、パイプラインが最後に実行されてから変更された行のみを処理するように ADF に指示します。 スキーマ ドリフトを使用した増分抽出を有効にするには、ネイティブ変更データ キャプチャが有効になっているテーブルではなく、Incremental/Watermark 列に基づくテーブルを選択します。

増分列: 増分抽出機能を使用する場合は、ソース テーブルのウォーターマークとして使用する日時または数値列を選択する必要があります。

ネイティブ変更データ キャプチャを有効にする (プレビュー): このオプションを使用して、パイプラインが最後に実行されてから SQL 変更データ キャプチャ テクノロジによってキャプチャされた差分データのみを処理するように ADF に指示します。 このオプションを使用すると、行の挿入、更新、削除を含む差分データが自動的に読み込まれ、増分列は必要ありません。 ADF でこのオプションを使用する前に、Azure SQL DB で変更データ キャプチャを有効にする必要があります。 ADF のこのオプションの詳細については、「ネイティブ変更データ キャプチャ」を参照してください。

最初から読み取りを開始する: 増分抽出でこのオプションを設定すると、増分抽出が有効になっているパイプラインの最初の実行時にすべての行を読み取るよう ADF に指示します。

シンク変換

Azure SQL Database に固有の設定は、シンク変換の [設定] タブにあります。

[Update method](更新方法) :対象となるデータベースに対して許可される操作を指定します。 既定では、挿入のみが許可されます。 行を更新、アップサート、または削除するには、それらのアクションに対して行をタグ付けするために行の変更変換が必要になります。 更新、アップサート、削除の場合、1 つまたは複数のキー列を設定して、変更する行を決定する必要があります。

ここでキーとして選択する列の名前は、後続の更新、アップサート、削除でサービスによって使用されます。 そのため、シンク マッピングに存在する列を選択する必要があります。 このキー列に値を書き込まない場合、[Skip writing key columns](キー列の書き込みをスキップする) をクリックします。

ここで使用するキー列は、ターゲットの Azure SQL Database テーブルを更新するためにパラメーター化できます。 複合キーの複数の列がある場合、[カスタム式] をクリックすると、データ フロー式言語を使用して動的コンテンツを追加できます。これには、複合キーの列名を含む文字列の配列を含めることができます。

[Table action](テーブル アクション): 書き込み前に変換先テーブルのすべての行を再作成するか削除するかを指定します。

• [なし]:テーブルに対してアクションは実行されません。
• [再作成]:テーブルが削除され、再作成されます。 新しいテーブルを動的に作成する場合に必要です。
• [切り詰め]:ターゲット テーブルのすべての行が削除されます。

Batch size: 各バケットに書き込まれる行数を制御します。 バッチ サイズを大きくすると、圧縮とメモリの最適化が向上しますが、データをキャッシュする際にメモリ不足の例外が発生するリスクがあります。

[Use TempDB](TempDB を使用): 既定では、読み込みプロセスの一環としてデータを保存するために、グローバル一時テーブルが使用されます。 [Use TempDB](TempDB を使用) オプションをオフにし、代わりに、このシンクに使用されているデータベース内にあるユーザー データベースに、一時的に保持するテーブルを保存するようサービスに要求することもできます。

[Pre and Post SQL scripts] (事前および事後 SQL スクリプト) : データがシンク データベースに書き込まれる前 (前処理) と書き込まれた後 (後処理) に実行される複数行の SQL スクリプトを入力します。

エラー行の処理

Azure SQL DB に書き込む場合、書き込み先で設定されている制約によって、データの特定の行が失敗することがあります。 一般的なエラーには次のようなものがあります。

• テーブル内の文字列データまたはバイナリ データが切り捨てられる
• 列に値 NULL を挿入できない
• INSERT ステートメントが CHECK 制約と競合している

既定では、データ フローの実行は最初に発生したエラーで失敗します。 [エラーのまま続行する] を選択すると、個々の行でエラーが発生した場合でもデータ フローを完了することができます。 サービスには、これらのエラー行を処理するためのさまざまなオプションが用意されています。

[Transaction Commit](トランザクション コミット): データを 1 つのトランザクションまたはバッチのどちらで書き込むかを選択します。 1 つのトランザクションの場合はパフォーマンスが低下しますが、トランザクションが完了するまで書き込みデータは他のユーザーに表示されません。

[Output rejected data](拒否されたデータの出力): 有効にすると、エラー行を Azure Blob Storage または選択した Azure Data Lake Storage Gen2 アカウントの csv ファイルに出力できます。 これにより、3 つの列 (INSERT または UPDATE などの SQL 操作、データ フロー エラー コード、および行のエラー メッセージ) を含むエラー行が書き込まれます。

[Report success on error](エラー発生時に成功を報告): 有効にすると、エラー行が見つかった場合でもデータ フローは成功としてマークされます。

Azure SQL Database のデータ型のマッピング

Azure SQL Database をコピー元またはコピー先としてデータがコピーされると、次の Azure SQL Database のデータ型から Azure Data Factory の中間データ型へのマッピングが使用されます。 Azure Data Factory を直接実装する Synapse パイプライン機能でも同じマッピングが使用されます。 コピー アクティビティでソースのスキーマとデータ型がシンクにマッピングされるしくみについては、スキーマとデータ型のマッピングに関する記事を参照してください。

Lookup アクティビティのプロパティ

プロパティの詳細については、Lookup アクティビティに関するページを参照してください。

GetMetadata アクティビティのプロパティ

プロパティの詳細については、GetMetadata アクティビティに関するページを参照してください。

Always Encrypted の使用

Always Encrypted を使用して Azure SQL Database との間でデータをコピーする場合は、次の手順に従います。

1. 列マスター キー (CMK) を Azure Key Vault に保存します。 詳細については、Azure Key Vault を使用して Always Encrypted を構成する方法に関する記事を参照してください

2. 列マスター キー (CMK) が格納されているキー コンテナーへのアクセス権を取得します。 必要なアクセス許可については、こちらの記事を参照してください。

3. リンク サービスを作成して SQL データベースに接続し、マネージド ID またはサービス プリンシパルを使用して "Always Encrypted" 機能を有効にします。

ネイティブ変更データ キャプチャ

Azure Data Factory は、SQL Server、Azure SQL DB、Azure SQL MI のネイティブ変更データ キャプチャ機能をサポートできます。 SQL ストアの行の挿入、更新、削除などの変更されたデータは、ADF マッピング データフローによって自動的に検出および抽出できます。 マッピング データフローのコード エクスペリエンスがないため、ユーザーはデータベースを宛先のストアとして追加することで、SQL ストアからのデータ レプリケーション シナリオを簡単に実現できます。 さらに、ユーザーは間にデータ変換ロジックを作成して、SQL ストアから増分 ETL シナリオを実現することもできます。

チェックポイントを ADF が記録して最後の実行から変更データを自動的に取得できるようにするために、パイプラインとアクティビティ名は変更しないようにしてください。 パイプライン名またはアクティビティ名を変更すると、チェックポイントがリセットされます。これにより、次回の実行時に最初から変更を開始したり、変更を取得したりすることができます。 パイプライン名またはアクティビティ名を変更しても、直近の実行の後に変更されたデータを自動的に取得するためにチェックポイントを保持する場合は、データフロー アクティビティで独自のチェックポイント キーを使用して、これを実現してください。

パイプラインをデバッグすると、この機能は同じように動作します。 デバッグ実行中にブラウザーを更新すると、チェックポイントがリセットされることに注意してください。 デバッグ実行のパイプライン結果に問題がなければ、パイプラインの発行とトリガーに進むことができます。 最初に発行されたパイプラインをトリガーした時点では、最初から自動的に再起動されるか、またはその後からの変更が取得されます。

[モニター] セクションでは、常にパイプラインを再実行できます。 この場合、変更されたデータは、選択したパイプライン実行の前のチェックポイントから常にキャプチャされます。

例 1:

マッピング データフロー内のデータベースに参照されるシンク変換を使用して SQL CDC 対応データセットに参照されるソース変換を直接チェーンすると、SQL ソースで行われた変更がターゲット データベースに自動的に適用されるため、データベース間のデータ レプリケーション シナリオを簡単に取得できます。 シンク変換で更新方法を使用して、挿入を許可するか、更新を許可するか、ターゲット データベースで削除を許可するかを選択できます。 マッピング データフローのスクリプトの例を次に示します。

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 sink(allowSchemaDrift: true,
	validateSchema: false,
	deletable:true,
	insertable:true,
	updateable:true,
	upsertable:true,
	keys:['id'],
	format: 'table',
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true,
	errorHandlingOption: 'stopOnFirstError') ~> sink1

例 2:

SQL CDC を使用したデータベース間のデータ レプリケーションではなく ETL シナリオを有効にする場合は、isInsert(1)、isUpdate(1)、isDelete(1) などのマッピング データフローで式を使用して、異なる操作の種類の行を区別できます。 値を持つ 1 つの列を派生する際にデータフローをマッピングするためのスクリプト例の 1 つを次に示します。1 は挿入された行を示し、2 は更新された行を示し、3 はダウンストリーム変換で差分データを処理するために削除された行を示します。

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 derive(operationType = iif(isInsert(1), 1, iif(isUpdate(1), 2, 3))) ~> derivedColumn1
derivedColumn1 sink(allowSchemaDrift: true,
	validateSchema: false,
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> sink1

既知の制限事項:

• SQL CDC からの net 変更のみが、cdc.fn_cdc_get_net_changes_ 経由で ADF によって読み込まれます。

Azure SQL Database のバージョンをアップグレードする

Azure SQL Database のバージョンをアップグレードするには、[リンク サービスの編集] ページで [バージョン] の [推奨] を選択し、推奨バージョンのリンク サービス プロパティを参照してリンク サービスを構成します。

推奨バージョンとレガシ バージョンの違い

以下の表は、推奨バージョンとレガシ バージョンを使用する Azure SQL Database の違いを示したものです。

関連するコンテンツ

Copy アクティビティでソースおよびシンクとしてサポートされるデータ ストアの一覧については、「サポートされるデータ ストアと形式」を参照してください。