Azure Data Factory または Synapse Analytics を使用して Azure SQL Managed Instance のデータをコピーおよび変換する
適用対象:
Azure Data Factory Azure Synapse Analytics
ヒント
企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータ サイエンス、リアルタイム分析、ビジネス インテリジェンス、レポートまで、あらゆるものをカバーしています。 無料で新しい試用版を開始する方法について説明します。
この記事では、コピー アクティビティを使用して、Azure SQL Managed Instance との間でデータをコピーする方法、および Data Flow を使用して Azure SQL Managed Instance のデータを変換する方法について説明します。 詳細については、Azure Data Factory および Synapse Analytics の概要記事を参照してください。
サポートされる機能
この Azure SQL Managed Instance コネクタは、次の機能についてサポートされます。
| サポートされる機能 | IR | マネージド プライベート エンドポイント |
|---|---|---|
| Copy アクティビティ (ソース/シンク) | ① ② | ✓ パブリック プレビュー |
| マッピング データ フロー (ソース/シンク) | ① | ✓ パブリック プレビュー |
| Lookup アクティビティ | ① ② | ✓ パブリック プレビュー |
| GetMetadata アクティビティ | ① ② | ✓ パブリック プレビュー |
| スクリプト活動 | ① ② | ✓ パブリック プレビュー |
| ストアド プロシージャ アクティビティ | ① ② | ✓ パブリック プレビュー |
① Azure 統合ランタイム ② セルフホステッド統合ランタイム
コピー アクティビティについて、この Azure SQL Database コネクタは以下の機能をサポートします。
- SQL 認証および Microsoft Entra アプリケーション トークン認証と、Azure リソースのサービス プリンシパルまたはマネージド ID を使って、データをコピーする。
- ソースとして、SQL クエリまたはストアド プロシージャを使用してデータを取得する。 SQL MI ソースからの並列コピーを選択することもできます。詳細については、SQL MI からの並列コピーに関するセクションを参照してください。
- シンクとして、ソース スキーマに基づいて、宛先テーブルが存在しない場合はこれを自動的に作成する。テーブルにデータを追加するか、コピー中にカスタム ロジックを使用してストアド プロシージャを呼び出す。
前提条件
SQL Managed Instance のパブリック エンドポイントにアクセスするために、マネージド Azure Integration Runtime を使用できます。 パブリック エンドポイントを有効にするだけでなく、ネットワーク セキュリティ グループでのパブリック エンドポイント トラフィックも許可して、サービスが確実にデータベースに接続できるようにしてください。 詳細については、このガイダンスを参照してください。
SQL Managed Instance のプライベート エンドポイントにアクセスするには、そのデータベースにアクセスできるセルフホステッド統合ランタイムを設定します。 セルフホステッド統合ランタイムをマネージド インスタンスと同じ仮想ネットワーク内でプロビジョニングする場合は、統合ランタイム コンピューターがマネージド インスタンスとは別のサブネットにあることを確認してください。 セルフホステッド統合ランタイムをマネージド インスタンスとは別の仮想ネットワークにプロビジョニングする場合は、仮想ネットワーク ピアリングまたは仮想ネットワーク間接続のどちらかを使用できます。 詳細については、SQL Managed Instance へのアプリケーションの接続に関するページを参照してください。
はじめに
パイプラインでコピー アクティビティを実行するには、次のいずれかのツールまたは SDK を使用します。
UI を使用して Azure SQL Managed Instance のリンク サービスを作成する
次の手順を使用して、Azure portal UI で SQL Managed Instance のリンク サービスを作成します。
Azure Data Factory または Synapse ワークスペースの [管理] タブに移動し、[リンクされたサービス] を選択して、[新規] をクリックします。
SQL を検索し、Azure SQL Server Managed Instance コネクタを選択します。
サービスの詳細を構成し、接続をテストして、新しいリンク サービスを作成します。
コネクタの構成の詳細
以降のセクションでは、SQL Managed Instance コネクタに固有の Azure Data Factory エンティティを定義するために使用されるプロパティについて詳しく説明します。
リンクされたサービスのプロパティ
SQL Managed Instance のリンクされたサービスでは、次の汎用プロパティがサポートされます。
| プロパティ | Description | 必須 |
|---|---|---|
| type | type プロパティを AzureSqlMI に設定する必要があります。 | はい |
| connectionString | このプロパティでは、SQL 認証を使用して SQL Managed Instance に接続するために必要な connectionString 情報を指定します。 詳細については、次の例を参照してください。 既定のポートは 1433 です。 パブリック エンドポイントを持つ SQL Managed Instance を使用する場合は、ポート 3342 を明示的に指定します。 また、Azure Key Vault にパスワードを格納することもできます。 それが SQL 認証である場合は、接続文字列から password 構成を取得します。 詳細については、この表の後にある JSON の例および「Azure Key Vault への資格情報の格納」を参照してください。 |
はい |
| azureCloudType | サービス プリンシパル認証用に、Microsoft Entra アプリケーションが登録されている Azure クラウド環境の種類を指定します。 指定できる値は、AzurePublic、AzureChina、AzureUsGovernment、および AzureGermany です。 既定では、サービスのクラウド環境が使用されます。 |
いいえ |
| alwaysEncryptedSettings | マネージド ID またはサービス プリンシパルを使用して、SQL サーバーに格納されている機密データを保護する Always Encrypted を有効にするために必要な alwaysencryptedsettings 情報を指定します。 詳細については、この表の後にある JSON の例および「Always Encrypted の使用」を参照してください。 指定されていない場合、既定の always encrypted 設定は無効になります。 | いいえ |
| connectVia | この統合ランタイムは、データ ストアに接続するために使用されます。 マネージド インスタンスにパブリック エンドポイントがあり、サービスからそれにアクセスできるようにする場合は、セルフホステッド統合ランタイムまたは Azure 統合ランタイムを使用できます。 指定されていない場合は、既定の Azure Integration Runtime が使用されます。 | はい |
さまざまな認証の種類に対する固有のプロパティ、前提条件、および JSON サンプルについては、以降のセクションをご覧ください。
SQL 認証
SQL 認証の種類を使用するには、前のセクションで説明した汎用プロパティを指定します。
例 1: SQL 認証を使用する
{
"name": "AzureSqlMILinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;Password=<password>;"
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
例 2: Azure Key Vault 内のパスワードで SQL 認証を使用する
{
"name": "AzureSqlMILinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;",
"password": {
"type": "AzureKeyVaultSecret",
"store": {
"referenceName": "<Azure Key Vault linked service name>",
"type": "LinkedServiceReference"
},
"secretName": "<secretName>"
}
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
例 3: Always Encrypted で SQL 認証を使用する
{
"name": "AzureSqlMILinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;Password=<password>;"
},
"alwaysEncryptedSettings": {
"alwaysEncryptedAkvAuthType": "ServicePrincipal",
"servicePrincipalId": "<service principal id>",
"servicePrincipalKey": {
"type": "SecureString",
"value": "<service principal key>"
}
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
サービス プリンシパルの認証
前のセクションで説明した汎用的なプロパティに加えて、サービス プリンシパルの認証を使用するには、次のプロパティを指定します。
| プロパティ | Description | 必須 |
|---|---|---|
| servicePrincipalId | アプリケーションのクライアント ID を取得します。 | はい |
| servicePrincipalKey | アプリケーションのキーを取得します。 このフィールドを SecureString とマークして安全に保存するか、Azure Key Vault に保存されているシークレットを参照します。 | はい |
| tenant | ドメイン名やテナント ID など、アプリケーションが存在するテナントの情報を指定します。 これは、Azure portal の右上隅をマウスでポイントすることで取得できます。 | はい |
以下の手順に従う必要もあります。
Azure portal から Microsoft Entra アプリケーションを作成します。 アプリケーション名と、リンクされたサービスを定義する次の値を記録しておきます。
- アプリケーション ID
- アプリケーション キー
- テナント ID
サービス プリンシパルのログインを作成します。 SQL Server Management Studio (SSMS) で、sysadmin である SQL Server アカウントを使用して、マネージド インスタンスに接続します。 マスター データベースで、次の T-SQL を実行します。
CREATE LOGIN [your application name] FROM EXTERNAL PROVIDERサービス プリンシパルの包含データベース ユーザーを作成します。 データのコピー元またはコピー先のデータベースに接続し、次の T-SQL を実行します。
CREATE USER [your application name] FROM EXTERNAL PROVIDERSQL ユーザーおよび他のユーザーに対する通常の方法で、サービス プリンシパルに必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。
ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your application name]SQL Managed Instance のリンクされたサービスを構成します。
例: サービス プリンシパル認証を使用する
{
"name": "AzureSqlDbLinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;",
"servicePrincipalId": "<service principal id>",
"servicePrincipalKey": {
"type": "SecureString",
"value": "<service principal key>"
},
"tenant": "<tenant info, e.g. microsoft.onmicrosoft.com>"
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
システム割り当てマネージド ID 認証
データ ファクトリまたは Synapse ワークスペースは、他の Azure サービスに対して認証するためのサービスを表す、Azure リソースのシステム割り当てマネージド ID に関連付けることができます。 このマネージド ID を SQL Managed Instance の認証に使用できます。 この ID を使用して指定したサービスからデータベースにアクセスし、データベースとの間で双方向にデータをコピーできます。
システム割り当てマネージド ID 認証を使用するには、前のセクションで説明した汎用プロパティを指定し、次の手順に従います。
システム割り当てマネージド ID のログインを作成します。 SQL Server Management Studio (SSMS) で、sysadmin である SQL Server アカウントを使用して、マネージド インスタンスに接続します。 マスター データベースで、次の T-SQL を実行します。
CREATE LOGIN [your_factory_or_workspace_ name] FROM EXTERNAL PROVIDERシステム割り当てマネージド ID の包含データベース ユーザーを作成します。 データのコピー元またはコピー先のデータベースに接続し、次の T-SQL を実行します。
CREATE USER [your_factory_or_workspace_name] FROM EXTERNAL PROVIDER通常の SQL ユーザーなどと同様に、システム割り当てマネージド ID に必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。
ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your_factory_or_workspace_name]SQL Managed Instance のリンクされたサービスを構成します。
例: システム割り当てマネージド ID 認証を使用
{
"name": "AzureSqlDbLinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;"
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
ユーザー割り当てマネージド ID 認証
データ ファクトリまたは Synapse ワークスペースは、他の Azure サービスに対して認証するためのサービスを表す、ユーザー割り当てマネージド ID に関連付けることができます。 このマネージド ID を SQL Managed Instance の認証に使用できます。 この ID を使用して指定したサービスからデータベースにアクセスし、データベースとの間で双方向にデータをコピーできます。
前のセクションで説明した汎用的なプロパティに加えて、ユーザー割り当てマネージド ID 認証を使用するには、次のプロパティを指定します。
| プロパティ | Description | 必須 |
|---|---|---|
| 資格情報 | ユーザー割り当てマネージド ID を資格情報オブジェクトとして指定します。 | はい |
以下の手順に従う必要もあります。
ユーザー割り当てマネージド ID のログインを作成します。 SQL Server Management Studio (SSMS) で、sysadmin である SQL Server アカウントを使用して、マネージド インスタンスに接続します。 マスター データベースで、次の T-SQL を実行します。
CREATE LOGIN [your_factory_or_workspace_ name] FROM EXTERNAL PROVIDERユーザー割り当てマネージド ID の包含データベース ユーザーを作成します。 データのコピー元またはコピー先のデータベースに接続し、次の T-SQL を実行します。
CREATE USER [your_factory_or_workspace_name] FROM EXTERNAL PROVIDER1 つ以上のユーザー割り当てマネージド ID を作成し、通常の SQL ユーザーなどと同様に、ユーザー割り当てマネージド ID に必要なアクセス許可を付与します。 次のコードを実行します。 詳細については、こちらのドキュメントを参照してください。
ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your_factory_or_workspace_name]1 つ以上のユーザー割り当てマネージド ID をデータ ファクトリに割り当てて、ユーザー割り当てマネージド ID ごとに資格情報を作成します。
SQL Managed Instance のリンクされたサービスを構成します。
例: ユーザー割り当てマネージド ID 認証を使用
{
"name": "AzureSqlDbLinkedService",
"properties": {
"type": "AzureSqlMI",
"typeProperties": {
"connectionString": "Data Source=<hostname,port>;Initial Catalog=<databasename>;",
"credential": {
"referenceName": "credential1",
"type": "CredentialReference"
}
},
"connectVia": {
"referenceName": "<name of Integration Runtime>",
"type": "IntegrationRuntimeReference"
}
}
}
データセットのプロパティ
データセットを定義するために使用できるセクションとプロパティの完全な一覧については、データセットに関する記事をご覧ください。 このセクションでは、SQL Managed Instance データセットでサポートされるプロパティの一覧を示します。
SQL Managed Instance との間でデータをコピーするために、次のプロパティがサポートされています。
| プロパティ | Description | 必須 |
|---|---|---|
| type | データセットの type プロパティを AzureSqlMITable に設定する必要があります。 | はい |
| schema | スキーマの名前。 | ソースの場合はいいえ、シンクの場合ははい |
| table | テーブル/ビューの名前。 | ソースの場合はいいえ、シンクの場合ははい |
| tableName | スキーマがあるテーブル/ビューの名前。 このプロパティは下位互換性のためにサポートされています。 新しいワークロードでは、schema と table を使用します。 |
ソースの場合はいいえ、シンクの場合ははい |
例
{
"name": "AzureSqlMIDataset",
"properties":
{
"type": "AzureSqlMITable",
"linkedServiceName": {
"referenceName": "<SQL Managed Instance linked service name>",
"type": "LinkedServiceReference"
},
"schema": [ < physical schema, optional, retrievable during authoring > ],
"typeProperties": {
"schema": "<schema_name>",
"table": "<table_name>"
}
}
}
コピー アクティビティのプロパティ
アクティビティの定義に利用できるセクションとプロパティの完全な一覧については、パイプラインに関する記事を参照してください。 このセクションでは、SQL Managed Instance のソースとシンクでサポートされるプロパティの一覧を示します。
ソースとしての SQL Managed Instance
ヒント
データ パーティション分割を使用して、SQL MI からデータを効率的に読み込む方法の詳細については、SQL MI からの並列コピーに関するセクションを参照してください。
SQL Managed Instance からデータをコピーするために、コピー アクティビティ ソース セクションでは次のプロパティがサポートされています。
| プロパティ | Description | 必須 |
|---|---|---|
| type | コピー アクティビティのソースの type プロパティを SqlMISource に設定する必要があります。 | はい |
| sqlReaderQuery | このプロパティは、カスタム SQL クエリを使用してデータを読み取ります。 たとえば select * from MyTable です。 |
いいえ |
| sqlReaderStoredProcedureName | このプロパティは、ソース テーブルからデータを読み取るストアド プロシージャの名前です。 最後の SQL ステートメントはストアド プロシージャの SELECT ステートメントにする必要があります。 | いいえ |
| storedProcedureParameters | これらのパラメーターは、ストアド プロシージャ用です。 使用可能な値は、名前または値のペアです。 パラメーターの名前とその大文字と小文字は、ストアド プロシージャのパラメーターの名前とその大文字小文字と一致する必要があります。 |
いいえ |
| isolationLevel | SQL ソースのトランザクション ロック動作を指定します。 使用できる値は、次のとおりです。ReadCommitted、ReadUncommitted、RepeatableRead、Serializable、Snapshot。 指定しなかった場合は、データベースの既定の分離レベルが使用されます。 詳細についてはこちらのドキュメントをご覧ください。 | いいえ |
| partitionOptions | SQL MI からデータを読み込むときに使用するデータ パーティション分割オプションを指定します。 使用できる値は、以下のとおりです。None (既定値)、PhysicalPartitionsOfTable、および DynamicRange。 パーティション オプションが有効になっている (つまり、 None ではない) 場合、SQL MI から同時にデータを読み込む並列処理の次数は、コピー アクティビティの parallelCopies の設定によって制御されます。 |
いいえ |
| partitionSettings | データ パーティション分割の設定のグループを指定します。 パーティション オプションが None でない場合に適用されます。 |
いいえ |
partitionSettings の下: |
||
| partitionColumnName | 並列コピーの範囲パーティション分割で使用される整数型または日付/日時型 (int、smallint、bigint、date、smalldatetime、datetime、datetime2、または datetimeoffset) のソース列の名前を指定します。 指定されない場合は、テーブルのインデックスまたは主キーが自動検出され、パーティション列として使用されます。パーティション オプションが DynamicRange である場合に適用されます。 クエリを使用してソース データを取得する場合は、WHERE 句で ?AdfDynamicRangePartitionCondition をフックします。 例については、「SQL データベースからの並列コピー」セクションを参照してください。 |
いいえ |
| partitionUpperBound | パーティション範囲の分割のための、パーティション列の最大値。 この値は、テーブル内の行のフィルター処理用ではなく、パーティションのストライドを決定するために使用されます。 テーブルまたはクエリ結果に含まれるすべての行がパーティション分割され、コピーされます。 指定されていない場合は、コピー アクティビティによって値が自動検出されます。 パーティション オプションが DynamicRange である場合に適用されます。 例については、「SQL データベースからの並列コピー」セクションを参照してください。 |
いいえ |
| partitionLowerBound | パーティション範囲の分割のための、パーティション列の最小値。 この値は、テーブル内の行のフィルター処理用ではなく、パーティションのストライドを決定するために使用されます。 テーブルまたはクエリ結果に含まれるすべての行がパーティション分割され、コピーされます。 指定されていない場合は、コピー アクティビティによって値が自動検出されます。 パーティション オプションが DynamicRange である場合に適用されます。 例については、「SQL データベースからの並列コピー」セクションを参照してください。 |
いいえ |
以下の点に注意してください。
- SqlMISource に sqlReaderQuery が指定されている場合、コピー アクティビティでは、データを取得するために SQL Managed Instance のインスタンス ソースに対してこのクエリを実行します。 sqlReaderStoredProcedureName と storedProcedureParameters を指定して、ストアド プロシージャを指定することもできます (ストアド プロシージャでパラメーターを使用する場合)。
- ソースのストアド プロシージャを使用してデータを取得する場合、異なるパラメーター値が渡されたときに別のスキーマを返すようにストアド プロシージャが設計されていると、UI からスキーマをインポートするときや、テーブルの自動作成を使用して SQL データベースにデータをコピーするときに、エラーが発生したり、予期しない結果になったりする可能性があります。
例:SQL クエリの使用
"activities":[
{
"name": "CopyFromAzureSqlMI",
"type": "Copy",
"inputs": [
{
"referenceName": "<SQL Managed Instance input dataset name>",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "<output dataset name>",
"type": "DatasetReference"
}
],
"typeProperties": {
"source": {
"type": "SqlMISource",
"sqlReaderQuery": "SELECT * FROM MyTable"
},
"sink": {
"type": "<sink type>"
}
}
}
]
例:ストアド プロシージャの使用
"activities":[
{
"name": "CopyFromAzureSqlMI",
"type": "Copy",
"inputs": [
{
"referenceName": "<SQL Managed Instance input dataset name>",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "<output dataset name>",
"type": "DatasetReference"
}
],
"typeProperties": {
"source": {
"type": "SqlMISource",
"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 Managed Instance
ヒント
サポートされる書き込み動作、構成、およびベスト プラクティスの詳細については、「SQL Managed Instance にデータを読み込む場合のベスト プラクティス」を参照してください。
データを SQL Managed Instance にコピーするために、コピー アクティビティ シンク セクションでは次のプロパティがサポートされています。
| プロパティ | Description | 必須 |
|---|---|---|
| type | コピー アクティビティのシンクの type プロパティは、SqlMISink に設定する必要があります。 | はい |
| preCopyScript | このプロパティでは、コピー アクティビティで SQL Managed Instance にデータを書き込む前に実行する SQL クエリを指定します。 これは、コピー実行ごとに 1 回だけ呼び出されます。 このプロパティを使用して、事前に読み込まれたデータをクリーンアップできます。 | いいえ |
| tableOption | ソースのスキーマに基づいて自動的にシンク テーブルを作成するかどうかを指定します (存在しない場合)。 シンクでストアド プロシージャが指定されている場合、テーブルの自動作成はサポートされません。 使用できる値は none (既定値)、autoCreate です。 |
いいえ |
| sqlWriterStoredProcedureName | ターゲット テーブルにソース データを適用する方法を定義しているストアド プロシージャの名前です。 このストアド プロシージャはバッチごとに呼び出されます。 1 回だけ実行され、ソース データとは関係がない操作 (削除/切り詰めなど) の場合は、 preCopyScript プロパティを使用します。例については、「SQL シンクからのストアド プロシージャの呼び出し」を参照してください。 |
いいえ |
| storedProcedureTableTypeParameterName | ストアド プロシージャで指定されたテーブル型のパラメーター名。 | いいえ |
| sqlWriterTableType | ストアド プロシージャで使用するテーブル型の名前。 コピー アクティビティでは、このテーブル型の一時テーブルでデータを移動できます。 その後、ストアド プロシージャのコードにより、コピーされたデータを既存のデータと結合できます。 | いいえ |
| storedProcedureParameters | ストアド プロシージャのパラメーター。 使用可能な値は、名前と値のペアです。 パラメーターの名前とその大文字と小文字は、ストアド プロシージャのパラメーターの名前とその大文字小文字と一致する必要があります。 |
いいえ |
| writeBatchSize | SQL テーブルに挿入する "バッチあたりの" 行数。 使用可能な値は、行数の場合整数です。 既定では行のサイズに基づいて、サービスにより適切なバッチ サイズが動的に決定されます。 |
いいえ |
| writeBatchTimeout | このプロパティは、タイムアウトする前に一括挿入操作の完了を待つ時間を指定します。 使用可能な値は期間に対する値です。 たとえば "00:30:00" (30 分) を指定できます。 |
いいえ |
| maxConcurrentConnections | アクティビティの実行中にデータ ストアに対して確立されたコンカレント接続数の上限。 コンカレント接続を制限する場合にのみ、値を指定します。 | なし |
| WriteBehavior | Azure SQL MI にデータを読み込むコピー アクティビティの書き込み動作を指定します。 使用できる値は、Insert と Upsert です。 既定では、サービスは Insert を使用してデータを読み込みます。 |
いいえ |
| upsertSettings | 書き込み動作の設定のグループを指定します。 WriteBehavior オプションが Upsert である場合に適用します。 |
いいえ |
upsertSettings の下: |
||
| useTempDB | アップサートの中間テーブルとしてグローバル一時テーブルと物理テーブルのどちらを使用するかを指定します。 既定では、サービスはグローバル一時テーブルを中間テーブルとして使用します。 値は true です。 |
No |
| interimSchemaName | 物理テーブルを使う場合は、中間テーブルを作成するために中間スキーマを指定します。 注: ユーザーは、テーブルの作成と削除を行うアクセス許可を持っている必要があります。 既定では、中間テーブルはシンク テーブルと同じスキーマを共有します。 useTempDB オプションが False である場合に適用します。 |
No |
| キー | 行を一意に識別するための列名を指定します。 1 つのキーまたは一連のキーのいずれかを使用できます。 指定しない場合は、主キーが使用されます。 | いいえ |
例 1: データを追加する
"activities":[
{
"name": "CopyToAzureSqlMI",
"type": "Copy",
"inputs": [
{
"referenceName": "<input dataset name>",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "<SQL Managed Instance output dataset name>",
"type": "DatasetReference"
}
],
"typeProperties": {
"source": {
"type": "<source type>"
},
"sink": {
"type": "SqlMISink",
"tableOption": "autoCreate",
"writeBatchSize": 100000
}
}
}
]
例 2: コピー中にストアド プロシージャを呼び出す
詳細については、SQL MI シンクからのストアド プロシージャの呼び出しに関するセクションを参照してください。
"activities":[
{
"name": "CopyToAzureSqlMI",
"type": "Copy",
"inputs": [
{
"referenceName": "<input dataset name>",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "<SQL Managed Instance output dataset name>",
"type": "DatasetReference"
}
],
"typeProperties": {
"source": {
"type": "<source type>"
},
"sink": {
"type": "SqlMISink",
"sqlWriterStoredProcedureName": "CopyTestStoredProcedureWithParameters",
"storedProcedureTableTypeParameterName": "MyTable",
"sqlWriterTableType": "MyTableType",
"storedProcedureParameters": {
"identifier": { "value": "1", "type": "Int" },
"stringData": { "value": "str1" }
}
}
}
}
]
例 3: データをアップサートする
"activities":[
{
"name": "CopyToAzureSqlMI",
"type": "Copy",
"inputs": [
{
"referenceName": "<input dataset name>",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "<SQL Managed Instance output dataset name>",
"type": "DatasetReference"
}
],
"typeProperties": {
"source": {
"type": "<source type>"
},
"sink": {
"type": "SqlMISink",
"tableOption": "autoCreate",
"writeBehavior": "upsert",
"upsertSettings": {
"useTempDB": true,
"keys": [
"<column name>"
]
},
}
}
}
]
SQL MI からの並列コピー
コピー アクティビティのときに、Azure SQL Managed Instance コネクタには、データを並列でコピーする組み込みのデータ パーティション分割が用意されています。 データ パーティション分割オプションは、コピー アクティビティの [ソース] タブにあります。
パーティション分割でのコピーを有効にすると、コピー アクティビティによって自分の SQL MI ソースに対して並列クエリが実行され、パーティションごとにデータが読み込まれます。 並列度は、コピー アクティビティの parallelCopies 設定によって制御されます。 たとえば、parallelCopies を 4 に設定した場合、指定したパーティション オプションと設定に基づいて 4 つのクエリが同時に生成され、実行されます。各クエリでは、SQL MI からデータの一部を取得します。
特に、自分の SQL MI から大量のデータを読み込む場合は、データのパーティション分割を使用した並列コピーを有効にすることをお勧めします。 さまざまなシナリオの推奨構成を以下に示します。 ファイルベースのデータ ストアにデータをコピーする場合は、複数のファイルとしてフォルダーに書き込む (フォルダー名のみを指定する) ことをお勧めします。この場合、1 つのファイルに書き込むよりもパフォーマンスが優れています。
| シナリオ | 推奨設定 |
|---|---|
| 物理パーティションに分割された大きなテーブル全体から読み込む。 | パーティション オプション: テーブルの物理パーティション。 実行中に、サービスによって物理パーティションが自動的に検出され、パーティションごとにデータがコピーされます。 テーブルに物理パーティションがあるかどうかを確認するには、こちらのクエリを参照してください。 |
| 物理パーティションがなく、データ パーティション分割用の整数または日時の列がある大きなテーブル全体から読み込む。 | パーティション オプション: 動的範囲パーティション。 パーティション列 (省略可能):データのパーティション分割に使用される列を指定します。 指定されていない場合は、インデックスまたは主キー列が使用されます。 パーティションの上限とパーティションの下限 (省略可能):パーティションのストライドを決定する場合に指定します。 これは、テーブル内の行のフィルター処理用ではなく、テーブル内のすべての行がパーティション分割されてコピーされます。 指定されていない場合は、コピー アクティビティによって値が自動検出されます。 たとえば、パーティション列「ID」の値の範囲が 1 ～ 100 で、下限を 20 に、上限を 80 に設定し、並列コピーを 4 にした場合、サービスによって 4 つのパーティションでデータが取得されます。ID の範囲は、それぞれ、20 以下、21 ～ 50、51 ～ 80、81 以上となります。 |
| 物理パーティションがなく、データ パーティション分割用の整数列または日付/日時列がある大量のデータを、カスタム クエリを使用して読み込む。 | パーティション オプション: 動的範囲パーティション。 クエリ: SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>パーティション列: データのパーティション分割に使用される列を指定します。 パーティションの上限とパーティションの下限 (省略可能):パーティションのストライドを決定する場合に指定します。 これは、テーブル内の行のフィルター処理用ではなく、クエリ結果のすべての行がパーティション分割されてコピーされます。 指定しない場合、コピー アクティビティによって値が自動検出されます。 実行中に、 ?AdfRangePartitionColumnName が各パーティションの実際の列名および値の範囲に置き換えられ、SQL MI に送信されます。 たとえば、パーティション列「ID」の値の範囲が 1 ～ 100 で、下限を 20 に、上限を 80 に設定し、並列コピーを 4 にした場合、サービスによって 4 つのパーティションでデータが取得されます。ID の範囲は、それぞれ、20 以下、21 ～ 50、51 ～ 80、81 以上となります。 さまざまなシナリオのサンプル クエリを次に示します。 1.テーブル全体に対してクエリを実行する: SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition2.列の選択と追加の where 句フィルターが含まれるテーブルからのクエリ: SELECT <column_list> FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>3.サブクエリを使用したクエリ: SELECT <column_list> FROM (<your_sub_query>) AS T WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>4.サブクエリにパーティションがあるクエリ: SELECT <column_list> FROM (SELECT <your_sub_query_column_list> FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition) AS T |
パーティション オプションを使用してデータを読み込む場合のベスト プラクティス:
- データ スキューを回避するため、パーティション列 (主キーや一意キーなど) には特徴のある列を選択します。
- テーブルに組み込みパーティションがある場合は、パフォーマンスを向上させるためにパーティション オプションとして "テーブルの物理パーティション" を使用します。
- Azure Integration Runtime を使用してデータをコピーする場合は、より大きな (4 より大きい) "データ統合単位 (DIU)" (>4) を設定すると、より多くのコンピューティング リソースを利用できます。 そこで、該当するシナリオを確認してください。
- パーティション数は、"コピーの並列処理の次数" によって制御されます。この数値を大きくしすぎるとパフォーマンスが低下するため、この数値は、(DIU またはセルフホステッド IR ノードの数) x (2 から 4) に設定することをお勧めします。
例: 複数の物理パーティションがある大きなテーブル全体から読み込む
"source": {
"type": "SqlMISource",
"partitionOption": "PhysicalPartitionsOfTable"
}
例: 動的範囲パーティションを使用してクエリを実行する
"source": {
"type": "SqlMISource",
"query": "SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition 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 Managed Instance にデータを読み込む場合のベスト プラクティス
SQL Managed Instance にデータをコピーする場合は、さまざまな書き込み動作が必要になることがあります。
- 追加: ソース データには新しいレコードのみが含まれている。
- アップサート: ソース データには挿入と更新の両方が含まれている。
- 上書き: 毎回ディメンション テーブル全体を再度読み込みたい。
- カスタム ロジックでの書き込み: 宛先テーブルへの最終挿入の前に追加の処理が必要である。
構成方法とベスト プラクティスについては、対応するセクションを参照してください。
データを追加する
データの追加は、SQL Managed Instance シンク コネクタの既定の動作です。 サービスでは、テーブルに効率的に書き込むために一括挿入が実行されます。 コピー アクティビティで、それに応じてソースとシンクを構成できます。
データをアップサートする
Copy アクティビティは、データベースの一時テーブルにデータを読み込み、キーが存在する場合はシンク テーブルのデータを更新し、そうでない場合は新しいデータを挿入する処理をネイティブにサポートするようになりました。 コピー アクティビティのアップサート設定の詳細については、「シンクとしての SQL Managed Instance」を参照してください。
テーブル全体を上書きする
コピー アクティビティ シンクで preCopyScript プロパティを構成できます。 この場合、実行される Copy アクティビティごとに、サービスで最初にスクリプトが実行されます。 次に、コピーが実行されてデータが挿入されます。 たとえば、テーブル全体を最新のデータで上書きするには、ソースから新しいデータを一括で読み込む前に、すべてのレコードを最初に削除するスクリプトを指定します。
カスタム ロジックでデータを書き込む
カスタム ロジックでデータを書き込む手順は、「データをアップサートする」セクションで説明されている手順に似ています。 宛先テーブルへのソース データの最終挿入の前に追加の処理を適用する必要がある場合は、ステージング テーブルに読み込んだ後、ストアド プロシージャ アクティビティを呼び出すか、コピー アクティビティのシンクのストアド プロシージャを呼び出してデータを適用することができます。
SQL シンクからのストアド プロシージャの呼び出し
SQL Managed Instance にデータをコピーする場合は、ソース テーブルの各バッチに対して追加のパラメーターを使用して、ユーザー指定のストアド プロシージャを構成して呼び出すこともできます。 ストアド プロシージャ機能は テーブル値パラメーターを利用しています。
組み込みのコピー メカニズムでは目的を達成できない場合は、ストアド プロシージャを使用できます。 1 つの例は、宛先テーブルへのソース データの最終挿入の前に追加の処理を適用する場合です。 その他の処理の例をいくつか挙げると、列のマージ、追加の値の検索、複数のテーブルへの挿入があります。
次の例では、SQL Server データベース内のテーブルに upsert を行うストアド プロシージャを使用する方法を示します。 入力データとシンクの Marketing テーブルには、それぞれ 3 つの列 (ProfileID、State、Category) があるものとします。 ProfileID 列に基づいてアップサートを行い、"ProductA" という特定のカテゴリに対してのみ適用します。
データベースで、sqlWriterTableType と同じ名前のテーブル型を定義します。 テーブル型のスキーマは、入力データから返されるスキーマと同じです。
CREATE TYPE [dbo].[MarketingType] AS TABLE( [ProfileID] [varchar](256) NOT NULL, [State] [varchar](256) NOT NULL, [Category] [varchar](256) NOT NULL )データベース内で、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パイプラインで、コピー アクティビティの SQL MI シンク セクションを次のように定義します。
"sink": { "type": "SqlMISink", "sqlWriterStoredProcedureName": "spOverwriteMarketing", "storedProcedureTableTypeParameterName": "Marketing", "sqlWriterTableType": "MarketingType", "storedProcedureParameters": { "category": { "value": "ProductA" } } }
Mapping Data Flow のプロパティ
マッピング データ フローでデータを変換する場合、Azure SQL Managed Instance からテーブルの読み取りと書き込みを実行できます。 詳細については、マッピング データ フローのソース変換とシンク変換に関する記事をご覧ください。
ソース変換
次の表に、Azure SQL Managed Instance ソースでサポートされるプロパティの一覧を示します。 これらのプロパティは、 [ソース オプション] タブで編集できます。
| 名前 | 説明 | 必須 | 使用できる値 | データ フロー スクリプトのプロパティ |
|---|---|---|---|---|
| テーブル | [テーブル] を入力として選択した場合、データセットで指定されたテーブルからすべてのデータがデータ フローによってフェッチされます。 | いいえ | - | - |
| クエリ | [クエリ] を入力として選択した場合は、ソースからデータをフェッチする SQL クエリを指定します。これにより、データセットで指定したテーブルがオーバーライドされます。 テストまたはルックアップ対象の行を減らすうえで、クエリの使用は有効な手段です。 Order By 句はサポートされていませんが、完全な SELECT FROM ステートメントを設定することができます。 ユーザー定義のテーブル関数を使用することもできます。 select * from udfGetData() は、データ フローで使用できるテーブルを返す SQL の UDF です。 クエリの例: Select * from MyTable where customerId > 1000 and customerId < 2000 |
いいえ | String | query |
| バッチ サイズ | 大量データを読み取りにまとめるバッチ サイズを指定します。 | いいえ | Integer | batchSize |
| Isolation Level | 次のいずれかの分離レベルを選択します。 - コミットされたものを読み取り - コミットされていないものを読み取り (既定値) - 反復可能読み取り - シリアル化可能 - なし (分離レベルを無視) |
いいえ | READ_COMMITTED READ_UNCOMMITTED REPEATABLE_READ SERIALIZABLE NONE |
isolationLevel |
| 増分抽出を有効にする | このオプションを使用して、パイプラインが前回実行されてから変更された行のみを処理するように ADF に指示します。 | いいえ | - | - |
| 増分列 | 増分抽出機能を使用する場合は、ソース テーブルのウォーターマークとして使用する日時または数値列を選択する必要があります。 | いいえ | - | - |
| ネイティブ変更データ キャプチャを有効にする (プレビュー) | このオプションを使用して、パイプラインが前回実行されてから SQL 変更データ キャプチャ テクノロジによってキャプチャされた差分データのみを処理するように ADF に指示します。 このオプションを使用すると、行の挿入、更新、削除を含む差分データが自動的に読み込まれ、増分列は必要ありません。 ADF でこのオプションを使用する前に、Azure SQL MI で変更データ キャプチャを有効にする必要があります。 ADF のこのオプションの詳細については、「ネイティブ変更データ キャプチャ」を参照してください。 | いいえ | - | - |
| 最初から読み取りを開始する | 増分抽出でこのオプションを設定すると、増分抽出が有効になっているパイプラインの最初の実行時にすべての行を読み取るよう ADF に指示します。 | いいえ | - | - |
ヒント
SQL の共通テーブル式 (CTE) は、マッピング データ フローの Query モードではサポートされません。このモードを使用する前提条件は、クエリを SQL クエリの FROM 句で使用することはできるが CTE では使用できないという条件であるためです。 CTE を使用するには、次のクエリを使用してストアド プロシージャを作成する必要があります。
CREATE PROC CTESP @query nvarchar(max)
AS
BEGIN
EXECUTE sp_executesql @query;
END
次に、マッピング データ フローのソース変換でストアド プロシージャ モードを使用し、with CTE as (select 'test' as a) select * from CTE の例のように @query を設定します。 その後は期待どおりに CTE を使用できます。
Azure SQL Managed Instance ソース スクリプトの例
ソースの種類として Azure SQL Managed Instance を使用する場合、関連付けられているデータ フロー スクリプトは次のようになります。
source(allowSchemaDrift: true,
validateSchema: false,
isolationLevel: 'READ_UNCOMMITTED',
query: 'select * from MYTABLE',
format: 'query') ~> SQLMISource
シンク変換
次の表に、Azure SQL Managed Instance シンクでサポートされるプロパティの一覧を示します。 これらのプロパティは、 [シンク オプション] タブで編集できます。
| 名前 | 説明 | 必須 | 使用できる値 | データ フロー スクリプトのプロパティ |
|---|---|---|---|---|
| 更新方法 | 対象となるデータベースに対して許可される操作を指定します。 既定では、挿入のみが許可されます。 行を更新、アップサート、または削除するには、それらのアクションに対して行をタグ付けするために行の変更変換が必要になります。 |
はい | true または false |
deletable insertable updateable upsertable |
| [キー列] | 更新、upsert、削除の場合、キー列 (複数可) を設定して、変更する行を決定する必要があります。 キーとして選択する列の名前は、後続の更新、upsert、削除の一部として使用されます。 そのため、シンク マッピングに存在する列を選択する必要があります。 |
いいえ | Array | キー |
| Skip writing key columns\(キー列の書き込みをスキップする) | キー列に値を書き込まない場合は、[Skip writing key columns](キー列の書き込みをスキップする) を選択します。 | いいえ | true または false |
skipKeyWrites |
| テーブル アクション | 書き込み前に変換先テーブルのすべての行を再作成するか削除するかを指定します。 - なし: テーブルに対してアクションは実行されません。 - Recreate:テーブルが削除され、再作成されます。 新しいテーブルを動的に作成する場合に必要です。 - Truncate:ターゲット テーブルのすべての行が削除されます。 |
いいえ | true または false |
recreate truncate |
| バッチ サイズ | 各バッチで書き込まれる行の数を指定します。 バッチ サイズを大きくすると、圧縮とメモリの最適化が向上しますが、データをキャッシュする際にメモリ不足の例外が発生するリスクがあります。 | いいえ | Integer | batchSize |
| 事前および事後の SQL スクリプト | データがシンク データベースに書き込まれる前 (前処理) と書き込まれた後 (後処理) に実行される複数行の SQL スクリプトを指定します。 | いいえ | String | preSQLs postSQLs |
ヒント
- 複数のコマンドを含む単一のバッチ スクリプトを複数のバッチに分割することをお勧めします。
- バッチの一部として実行できるのは、単純に更新数を返すデータ操作言語 (DML) ステートメントおよびデータ定義言語 (DDL) ステートメントだけです。 詳細については、「バッチ操作の実行」を参照してください
Azure SQL Managed Instance シンク スクリプトの例
シンクの種類として Azure SQL Managed Instance を使用する場合、関連付けられているデータ フロー スクリプトは次のようになります。
IncomingStream sink(allowSchemaDrift: true,
validateSchema: false,
deletable:false,
insertable:true,
updateable:true,
upsertable:true,
keys:['keyColumn'],
format: 'table',
skipDuplicateMapInputs: true,
skipDuplicateMapOutputs: true) ~> SQLMISink
Lookup アクティビティのプロパティ
プロパティの詳細については、Lookup アクティビティに関するページを参照してください。
GetMetadata アクティビティのプロパティ
プロパティの詳細については、GetMetadata アクティビティに関するページを参照してください。
SQL Managed Instance のデータ型のマッピング
コピー アクティビティを使用して SQL Managed Instance との間でデータをコピーする場合、SQL Managed Instance のデータ型からサービス内で内部的に使用される中間データ型への、以下のマッピングが使用されます。 コピー アクティビティでソースのスキーマとデータ型がシンクにマッピングされるしくみについては、スキーマとデータ型のマッピングに関する記事を参照してください。
| SQL Managed Instance のデータ型 | 中間サービス データ型 |
|---|---|
| bigint | Int64 |
| binary | Byte[] |
| bit | Boolean |
| char | String, Char[] |
| date | DateTime |
| Datetime | DateTime |
| datetime2 | DateTime |
| Datetimeoffset | DateTimeOffset |
| Decimal | Decimal (10 進数型) |
| FILESTREAM attribute (varbinary(max)) | Byte[] |
| Float | Double |
| image | Byte[] |
| INT | Int32 |
| money | Decimal (10 進数型) |
| nchar | String, Char[] |
| ntext | String, Char[] |
| numeric | Decimal (10 進数型) |
| nvarchar | String, Char[] |
| real | Single |
| rowversion | Byte[] |
| smalldatetime | DateTime |
| smallint | Int16 |
| smallmoney | Decimal (10 進数型) |
| sql_variant | Object |
| text | String, Char[] |
| time | TimeSpan |
| timestamp | Byte[] |
| tinyint | Int16 |
| UNIQUEIDENTIFIER | Guid |
| varbinary | Byte[] |
| varchar | String, Char[] |
| xml | String |
注意
10 進の中間型にマップされるデータ型の場合、コピー アクティビティでは、現在、最大 28 の有効桁数がサポートされています。 28 よりも大きな有効桁数を必要とするデータがある場合は、SQL クエリで文字列に変換することを検討してください。
Always Encrypted の使用
Always Encrypted を使用して SQL Managed Instance との間でデータをコピーする場合は、次の手順に従います。
列マスター キー (CMK) を Azure Key Vault に保存します。 詳細については、Azure Key Vault を使用して Always Encrypted を構成する方法に関する記事を参照してください
列マスター キー (CMK) が格納されているキー コンテナーへのアクセス権を付与します。 必要なアクセス許可については、こちらの記事を参照してください。
リンク サービスを作成して SQL データベースに接続し、マネージド ID またはサービス プリンシパルを使用して "Always Encrypted" 機能を有効にします。
Note
SQL Managed Instance Always Encrypted は、次のシナリオをサポートしています。
- ソース データ ストアまたはシンク データ ストアのいずれかで、キー プロバイダー認証の種類としてマネージド ID またはサービス プリンシパルを使用する。
- ソース データ ストアとシンク データ ストアの両方で、キー プロバイダー認証の種類としてマネージド ID を使用する。
- ソース データ ストアとシンク データ ストアの両方で、キー プロバイダー認証の種類として同じサービス プリンシパルを使用する。
注意
現在、SQL Managed Instance 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 によって読み込まれます。
関連するコンテンツ
コピー アクティビティによってソース、シンクとしてサポートされるデータ ストアの一覧については、サポートされるデータ ストアに関するセクションを参照してください。