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

適用対象: Azure Data Factory Azure Synapse Analytics

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

サポートされる機能

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

(1) Azure統合ランタイム (2) セルフホステッド統合ランタイム

コピー アクティビティによってソースまたはシンクとしてサポートされるデータ ストアの一覧については、サポートされるデータ ストアに関する記事の表をご覧ください。

具体的には、このSQL Server コネクタは次をサポートします。

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

SQL Server Express LocalDB はサポートされていません。

前提条件

データ ストアがオンプレミス ネットワーク、Azure仮想ネットワーク、または Amazon Virtual Private Cloud 内にある場合は、自身がホストする統合ランタイム を構成して接続する必要があります。

データ ストアがマネージド クラウド データ サービスの場合は、Azure Integration Runtimeを使用できます。 アクセスがファイアウォール規則で承認されている IP に制限されている場合は、許可リストに Azure Integration Runtime IP を追加できます。

Azure Data Factoryの 管理された仮想ネットワーク統合ランタイム機能を使用して、セルフホステッド統合ランタイムをインストールして構成することなく、オンプレミス ネットワークにアクセスすることもできます。

Data Factory によってサポートされるネットワーク セキュリティ メカニズムやオプションの詳細については、「データ アクセス戦略」を参照してください。

概要

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

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

UI を使用してSQL Serverリンクされたサービスを作成する

Azure ポータル UI でSQL Serverリンクされたサービスを作成するには、次の手順に従います。

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

   • Azure Data Factory
   • Azure Synapse

   Azure Data Factory UI を使用した新しいリンク サービスの作成のスクリーンショット

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

   SQL Server コネクタのスクリーンショット

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

   

コネクタの構成の詳細

次のセクションでは、SQL Server データベース コネクタに固有の Data Factory および Synapse パイプライン エンティティの定義に使用されるプロパティについて詳しく説明します。

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

SQL Server Recommended バージョンでは TLS 1.3 がサポートされています。 Legacy バージョンを使用する場合は、このセクションを参照してSQL Serverリンクされたサービスをアップグレードしてください。 プロパティの詳細については、対応するセクションを参照してください。

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

推奨されるバージョン

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

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

SQL 認証

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

例: SQL 認証を使用する

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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 内のパスワードを使用して SQL 認証を行う

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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"
        }
    }
}

Windows 認証

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

例: Windows 認証を使用する

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

例: Azure Key Vault で Windows 認証をパスワードとともに使用する

{
    "name": "SqlServerLinkedService",
    "properties": {
        "annotations": [],
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

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

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

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

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

1. ユーザー割り当てマネージド ID にアクセス許可を付与します。

2. Microsoft Entra 認証を有効化するには、Azure VM 上の SQL Server を使用します。

3. ユーザー割り当てマネージド ID の包含データベース ユーザーを作成します。 少なくとも ALTER ANY USER 権限を持つMicrosoft Entra ID を使用して、SQL Server Management Studioなどのツールを使用して、データのコピー元またはコピー先のデータベースに接続します。 次の T-SQL を実行します。

   CREATE USER [your_resource_name] FROM EXTERNAL PROVIDER;
   

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

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

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

6. SQL Server のリンクされたサービスを構成します。

例

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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"
        }
    }
}

レガシ バージョン

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

この SQL Server コネクタでは、次の認証の種類がサポートされています。 詳細については、対応するセクションをご覧ください。

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

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

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

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

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

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

データセットを定義するために使用できるセクションとプロパティの完全な一覧については、データセットに関する記事をご覧ください。 このセクションでは、SQL Server データセットでサポートされるプロパティの一覧を示します。

SQL Server データベースとの間でデータをコピーするには、次のプロパティがサポートされています。

例

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

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

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

ソースとしてのSQL Server

SQL Serverからデータをコピーするには、コピー アクティビティのソースの種類を SqlSource に設定します。 コピー アクティビティの source セクションでは、次のプロパティがサポートされます。

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

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

例: SQL クエリを使う

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

例: ストアド プロシージャを使う

"activities":[
    {
        "name": "CopyFromSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Server input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlSource",
                "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

シンクとしてのSQL Server

データをSQL Serverにコピーするには、コピー アクティビティのシンクの種類を SqlSink に設定します。 コピー アクティビティの sink セクションでは、次のプロパティがサポートされます。

例 1: データを追加する

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

"source": {
    "type": "SqlSource",
    "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" と表示されます。

SQL Serverにデータを読み込むためのベスト プラクティス

SQL Serverにデータをコピーするときに、異なる書き込み動作が必要になる場合があります。

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

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

データを追加する

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

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

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

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

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

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

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

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

SQL Server データベースにデータをコピーする場合は、ソース テーブルの各バッチに追加パラメーターを指定して、ユーザー指定のストアド プロシージャを構成して呼び出すこともできます。 ストアド プロシージャ機能は テーブル値パラメーターを利用しています。 このサービスでは、ストアド プロシージャがそれ自体のトランザクションでラップされることにご留意ください。つまり、ストアド プロシージャ内で作成されたトランザクションはすべて入れ子になり、例外処理を意味することがあります。

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

次の例では、ストアド プロシージャを使用して、SQL Server データベース内のテーブルにアップサートを実行する方法を示します。 入力データと、シンクの 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. コピー アクティビティの SQL シンク セクションを次のように定義します。

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

Mapping Data Flow のプロパティ

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

ソース変換

次の表に、SQL Server ソースでサポートされるプロパティを示します。 これらのプロパティは、[ソース オプション] タブで編集できます。

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

SQL Server ソース スクリプトの例

ソースの種類としてSQL Serverを使用する場合、関連するデータ フロー スクリプトは次のようになります。

source(allowSchemaDrift: true,
    validateSchema: false,
    isolationLevel: 'READ_UNCOMMITTED',
    query: 'select * from MYTABLE',
    format: 'query') ~> SQLSource

シンク変換

次の表に、SQL Server のシンクでサポートされるプロパティを示します。 これらのプロパティは、[シンク オプション] タブで編集できます。

SQL Server シンク スクリプトの例

シンクの種類としてSQL Serverを使用する場合、関連付けられているデータ フロー スクリプトは次のようになります。

IncomingStream sink(allowSchemaDrift: true,
    validateSchema: false,
    deletable:false,
    insertable:true,
    updateable:true,
    upsertable:true,
    keys:['keyColumn'],
    format: 'table',
    skipDuplicateMapInputs: true,
    skipDuplicateMapOutputs: true) ~> SQLSink

SQL Serverのデータ型マッピング

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

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

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

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

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

Always Encrypted の使用

Always Encrypted、次の手順に従います。

1. Column マスター キー (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 によって読み込まれます。

接続の問題のトラブルシューティング

1. リモート接続を受け入れるようにSQL Server インスタンスを構成します。 SQL Server Management Studio を起動し、server を右クリックし、 Properties を選択します。 一覧から [接続] を選択し、[このサーバーへのリモート接続を許可する] チェック ボックスをオンにします。

   

   詳細な手順については、「remote access サーバー構成オプションの構成」をご覧ください。

2. SQL Server 構成マネージャーを開始します。 SQL Server ネットワーク構成で目的のインスタンスを展開し、MSSQLSERVER の プロトコル を選択します。 右側のウィンドウにプロトコルが表示されます。 [TCP/IP] を右クリックして [有効化] を選択し、TCP/IP を有効にします。

   

   TCP/IP プロトコルの有効化の詳細および別の方法については、「サーバー ネットワーク プロトコルの有効化または無効化」をご覧ください。

3. 同じウィンドウで、[TCP/IP] をダブルクリックして、[TCP/IP のプロパティ] ウィンドウを起動します。

4. [IP アドレス] タブに切り替えます。下にスクロールして、[IPAll] セクションを表示します。 [TCP ポート] を書き留めます。 既定値は 1433 です。

5. Windows Firewall の ルールを作成して、このポートを通じて受信トラフィックを許可する。

6. 接続の確認: 完全修飾名を使用してSQL Serverに接続するには、別のコンピューターからSQL Server Management Studioを使用して確認します。 たとえば "<machine>.<domain>.corp.<company>.com,1433" です。

SQL Server バージョンをアップグレードする

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

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

次の表は、推奨バージョンとレガシ バージョンを使用したSQL Serverの違いを示しています。

関連コンテンツ

コピー アクティビティによってソースおよびシンクとしてサポートされるデータ ストアの一覧については、サポートされるデータ ストアに関するセクションを参照してください。