PowerShell を使用して変更追跡情報を使用してAzure SQL DatabaseからAzure Blob Storageにデータを増分読み込みする

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

Data Factory in Microsoft Fabric は、よりシンプルなアーキテクチャ、組み込みの AI、および新機能を備えた次世代のAzure Data Factoryです。データ統合を初めて使用する場合は、Fabric Data Factory から始めます。既存の ADF ワークロードをFabricにアップグレードして、データサイエンス、リアルタイム分析、レポートの新機能にアクセスできます。

Fabric無料試用版を開始します。
Azure Data Factory から Microsoft Fabric の Data Factory へのアップグレード

このチュートリアルでは、change tracking 情報に基づいてAzure SQL DatabaseのソースデータベースのデルタデータをAzure BLOB ストレージに読み込むパイプラインを使用して、Azure データファクトリを作成します。

このチュートリアルでは、以下の手順を実行します。

ソースデータストアを準備します。
データファクトリを作成します。
リンクされたサービスを作成します。
ソース、シンク、変更追跡の各データセットを作成します。
フルコピーパイプラインを作成、実行、監視します。
ソーステーブルのデータを追加または更新します。
増分コピーパイプラインを作成、実行、監視します。

注意

Azure Az PowerShell モジュールを使用してAzureを操作することをお勧めします。作業を開始するには、「Install Azure PowerShellを参照してください。 Az PowerShell モジュールに移行する方法については、「AzRM から Azを参照してください。

概要

データ統合ソリューションでは、初回データ読み込みの後、増分データを読み込む手法が広く利用されています。ソースデータストアから特定の期間に変更されたデータを簡単に切り出すことができる場合もあります (LastModifyTime、CreationTime など)。一方、前回データを処理した時点からの差分データを明示的に特定する方法がない場合もあります。 Azure SQL DatabaseやSQL Serverなどのデータストアでサポートされる Change Tracking テクノロジを使用して、差分データを識別できます。このチュートリアルでは、SQL Change Tracking テクノロジでAzure Data Factoryを使用して、差分データをAzure SQL DatabaseからAzure Blob Storageに増分読み込みする方法について説明します。 SQL Change Tracking テクノロジの詳細については、「change tracking in SQL Server」を参照してください。

エンドツーエンドワークフロー

ここでは、Change Tracking テクノロジを使用してデータを増分読み込みする一般的なエンドツーエンドのワークフローステップを取り上げます。

注意

Azure SQL DatabaseとSQL Serverの両方で、Change Tracking テクノロジがサポートされています。このチュートリアルでは、ソースデータストアとしてAzure SQL Databaseを使用します。 SQL Server インスタンスを使用することもできます。

履歴データの初回読み込みを実行する (1 回実行)。
1. Azure SQL Databaseのソースデータベースで Change Tracking テクノロジを有効にします。
2. データベースから SYS_CHANGE_VERSION の初期値をベースラインとして取得し、変更済みデータをキャプチャします。
3. ソースデータベースから Azure BLOB ストレージに完全なデータを読み込みます。
スケジュールに従って差分データの増分読み込みを実行する (データの初回読み込み後に定期的に実行)。
1. 新旧の SYS_CHANGE_VERSION 値を取得します。
2. 差分データを読み込みます。これは、sys.change_tracking_tables から得られる変更済みの行 (2 つの SYS_CHANGE_VERSION 値の間にある行) の主キーとソーステーブル内のデータとを結合した後、その差分データをターゲットに移動することによって行います。
3. 次回の差分読み込みに備えて SYS_CHANGE_VERSION を更新します。

高レベルのソリューション

このチュートリアルでは、次の 2 つの操作を実行する 2 つのパイプラインを作成します。

Initial load:ソースデータストア (Azure SQL Database) からコピー先データストア (Azure Blob Storage) にデータ全体をコピーするコピーアクティビティを含むパイプラインを作成します。
増分読み込み: 次のアクティビティを含んだパイプラインを作成して定期的に実行します。
1. 2 つのルックアップアクティビティを作成して、Azure SQL Databaseから新旧のSYS_CHANGE_VERSIONを取得し、コピーアクティビティに渡します。
2. 1 つのコピーアクティビティを作成して、2 つのSYS_CHANGE_VERSION値の間に挿入/更新/削除されたデータをAzure SQL DatabaseからAzure Blob Storageにコピーします。
3. 次回のパイプライン実行に備えて SYS_CHANGE_VERSION の値を更新するストアドプロシージャアクティビティを 1 つ作成します。

Azure サブスクリプションをお持ちでない場合は、開始する前に free アカウントを作成してください。

前提条件

Azure PowerShell。 Azure PowerShell。
Azure SQL Database。数据ベースをソースデータストアとして使用します。 Azure SQL Databaseにデータベースがない場合は、 Azure SQL Databaseに関する記事を参照してください。
Azure Storage アカウント。 シンク データストアとして使用する BLOB ストレージです。 Azureストレージアカウントがない場合は、ストレージアカウントの作成に関する記事を参照してください。 adftutorial という名前のコンテナーを作成します。

データベースにデータソーステーブルを作成する

SQL Server Management Studio を起動し、SQL Database に接続します。
サーバーエクスプローラーで目的のデータベースを右クリックして [新しいクエリ] を選択します。

データベースに対して次の SQL コマンドを実行し、data_source_table という名前のテーブルをソースデータストアとして作成します。

create table data_source_table
(
    PersonID int NOT NULL,
    Name varchar(255),
    Age int
    PRIMARY KEY (PersonID)
);

INSERT INTO data_source_table
    (PersonID, Name, Age)
VALUES
    (1, 'aaaa', 21),
    (2, 'bbbb', 24),
    (3, 'cccc', 20),
    (4, 'dddd', 26),
    (5, 'eeee', 22);

次の SQL クエリを実行して、データベースとソーステーブル (data_source_table) の Change Tracking 機構を有効にします。
注意
- <your database name> は、data_source_table がある実際のデータベースの名前に置き換えてください。
- 現行の例では、変更済みのデータが 2 日間維持されます。変更済みデータを読み込む間隔を 3 日おき、またはそれ以上にした場合、変更済みデータの一部が読み込まれません。 CHANGE_RETENTION の数値を増やす必要があります。または、変更済みデータの読み込み間隔を必ず 2 日以内としてください。詳細については、「データベースの変更の追跡を有効にする」を参照してください。
```
ALTER DATABASE <your database name>
SET CHANGE_TRACKING = ON  
(CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)  

ALTER TABLE data_source_table
ENABLE CHANGE_TRACKING  
WITH (TRACK_COLUMNS_UPDATED = ON)
```

新しいテーブルを作成し、ChangeTracking_version を既定値で格納するために、次のクエリを実行します。

create table table_store_ChangeTracking_version
(
    TableName varchar(255),
    SYS_CHANGE_VERSION BIGINT,
);

DECLARE @ChangeTracking_version BIGINT
SET @ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION();  

INSERT INTO table_store_ChangeTracking_version
VALUES ('data_source_table', @ChangeTracking_version)

注意

SQL Database の変更追跡を有効にした後、データが変更されていなければ、変更追跡バージョンの値は 0 になります。

次のクエリを実行して、データベースにストアドプロシージャを作成します。このストアドプロシージャをパイプラインで呼び出すことによって、前の手順で作成したテーブルの変更追跡バージョンを更新します。
```
CREATE PROCEDURE Update_ChangeTracking_Version @CurrentTrackingVersion BIGINT, @TableName varchar(50)
AS

BEGIN

UPDATE table_store_ChangeTracking_version
SET [SYS_CHANGE_VERSION] = @CurrentTrackingVersion
WHERE [TableName] = @TableName

END    
```

Azure PowerShell

Azure PowerShell。

Data Factory の作成

後で PowerShell コマンドで使用できるように、リソースグループ名の変数を定義します。次のコマンドテキストを PowerShell にコピーし、Azure リソースグループの名前を二重引用符で囲んで指定し、コマンドを実行します。 (例: "adfrg")。
```
$resourceGroupName = "ADFTutorialResourceGroup";
```
リソースグループが既に存在する場合、上書きしないようお勧めします。 $resourceGroupName 変数に別の値を割り当てて、コマンドをもう一度実行します。
データファクトリの場所の変数を定義します。
```
$location = "East US"
```
Azure リソースグループを作成するには、次のコマンドを実行します。
```
New-AzResourceGroup $resourceGroupName $location
```
リソースグループが既に存在する場合、上書きしないようお勧めします。 $resourceGroupName 変数に別の値を割り当てて、コマンドをもう一度実行します。
データファクトリ名の変数を定義します。

重要

データファクトリ名は、グローバルに一意となるように更新してください。
```
$dataFactoryName = "IncCopyChgTrackingDF";
```
データファクトリを作成するには、次の Set-AzDataFactoryV2 コマンドレットを実行します。
```
Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location $location -Name $dataFactoryName
```

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

Azure データファクトリの名前はグローバルに一意である必要があります。次のエラーが発生した場合は、名前を変更してからもう一度実行してください。
```
The specified Data Factory name 'ADFIncCopyChangeTrackingTestFactory' is already in use. Data Factory names must be globally unique.
```
Data Factory インスタンスを作成するには、Azureへのログインに使用するユーザーアカウントが、Azure サブスクリプションの contributor ロールまたは owner ロール、または administrator である必要があります。
Data Factory が現在使用できるAzureリージョンの一覧については、次のページで目的のリージョンを選択し、Analytics を展開して、Data Factory: Products available by region> を探します。データファクトリで使用されるデータストア (Azure Storage、Azure SQL Databaseなど) とコンピューティング (HDInsight など) は、他のリージョンに存在できます。

リンクされたサービスを作成します

データストアおよびコンピューティングサービスをデータファクトリにリンクするには、リンクされたサービスをデータファクトリに作成します。このセクションでは、Azure SQL Databaseで、Azure Storage アカウントとデータベースへのリンクされたサービスを作成します。

Azure Storage リンクサービスを作成します。

この手順では、Azure Storage アカウントをデータファクトリにリンクします。

以下の内容を記述した AzureStorageLinkedService.json という名前の JSON ファイルを C:\ADFTutorials\IncCopyChangeTrackingTutorial フォルダー内に作成します (フォルダーがない場合は作成します)。ファイルを保存する前に、<accountName>、<accountKey> をAzure ストレージアカウントの名前とキーに置き換えます。
```
{
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}
```
Azure PowerShell で、C:\ADFTutorials\IncCopyChangeTrackingTutorial フォルダーに切り替えます。

Set-AzDataFactoryV2LinkedService コマンドレットを実行して、リンクされたサービス AzureStorageLinkedService を作成します。次の例では、ResourceGroupName パラメーターと DataFactoryName パラメーターの値を渡しています。

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureStorageLinkedService" -File ".\AzureStorageLinkedService.json"

出力例を次に示します。

LinkedServiceName : AzureStorageLinkedService
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureStorageLinkedService

Azure SQL Database のリンク先サービスを作成します。

この手順では、データベースをデータファクトリにリンクします。

C:\ADFTutorials\IncCopyChangeTrackingTutorial フォルダーに AzureSQLDatabaseLinkedService.json という名前の JSON ファイルを作成します。ファイルの保存前に、<your-server-name> と <your-database-name> は、お使いのサーバー名とデータベース名に置き換えてください。また、データファクトリのマネージド IDへのアクセスを許可するように Azure SQL Serverを構成する必要もあります。
```
{
"name": "AzureSqlDatabaseLinkedService",
"properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Server=tcp:<your-server-name>.database.windows.net,1433;Database=<your-database-name>;"
        },
        "authenticationType": "ManagedIdentity",
        "annotations": []
    }
}
```

Azure PowerShell で、Set-AzDataFactoryV2LinkedService コマンドレットを実行して、リンクされたサービス (AzureSQLDatabaseLinkedService を作成します。

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSQLDatabaseLinkedService" -File ".\AzureSQLDatabaseLinkedService.json"

出力例を次に示します。

LinkedServiceName : AzureSQLDatabaseLinkedService
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService

データセットを作成する

この手順では、データのコピー元とコピー先を表すデータセットを作成します。 SYS_CHANGE_VERSION の格納場所を表すデータセットを作成します。

ソースデータセットを作成する

この手順では、ソースデータを表すデータセットを作成します。

以下の内容を記述した SourceDataset.json という名前の JSON ファイルを同じフォルダー内に作成します。

{
    "name": "SourceDataset",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "data_source_table"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

Set-AzDataFactoryV2Dataset コマンドレットを実行して、データセットSourceDataset を作成します。

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SourceDataset" -File ".\SourceDataset.json"

このコマンドレットの出力例を次に示します。

DatasetName       : SourceDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

シンクデータセットを作成する

この手順では、ソースデータストアからコピーされたデータを表すデータセットを作成します。

以下の内容を記述した SinkDataset.json という名前の JSON ファイルを同じフォルダー内に作成します。
```
{
    "name": "SinkDataset",
    "properties": {
        "type": "AzureBlob",
        "typeProperties": {
            "folderPath": "adftutorial/incchgtracking",
            "fileName": "@CONCAT('Incremental-', pipeline().RunId, '.txt')",
            "format": {
                "type": "TextFormat"
            }
        },
        "linkedServiceName": {
            "referenceName": "AzureStorageLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}
```
前提条件の一部として、Azure Blob Storageに adftutorial コンテナーを作成します。このコンテナーが存在しない場合は作成するか、または既存のコンテナーの名前に設定してください。このチュートリアルでは、@CONCAT('Incremental-', pipeline().RunId, '.txt') という式を使って出力ファイル名が動的に生成されます。

Set-AzDataFactoryV2Dataset コマンドレットを実行して、データセットSinkDataset を作成します。

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "SinkDataset" -File ".\SinkDataset.json"

このコマンドレットの出力例を次に示します。

DatasetName       : SinkDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureBlobDataset

変更追跡データセットを作成する

この手順では、変更追跡バージョンを格納するためのデータセットを作成します。

以下の内容を記述した ChangeTrackingDataset.json という名前の JSON ファイルを同じフォルダー内に作成します。

{
    "name": " ChangeTrackingDataset",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "table_store_ChangeTracking_version"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

前提条件の 1 つとして table_store_ChangeTracking_version テーブルを作成します。

Set-AzDataFactoryV2Dataset コマンドレットを実行して、データセット「ChangeTrackingDataset」を作成します。

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "ChangeTrackingDataset" -File ".\ChangeTrackingDataset.json"

このコマンドレットの出力例を次に示します。

DatasetName       : ChangeTrackingDataset
ResourceGroupName : ADFTutorialResourceGroup
DataFactoryName   : IncCopyChgTrackingDF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

完全コピーのパイプラインを作成する

この手順では、ソースデータストア (Azure SQL Database) からコピー先データストア (Azure Blob Storage) にデータ全体をコピーするコピーアクティビティを含むパイプラインを作成します。

JSON ファイル FullCopyPipeline.json を同じフォルダー内に作成します。次の内容を記述してください。

{
    "name": "FullCopyPipeline",
    "properties": {
        "activities": [{
            "name": "FullCopyActivity",
            "type": "Copy",
            "typeProperties": {
                "source": {
                    "type": "SqlSource"
                },
                "sink": {
                    "type": "BlobSink"
                }
            },

            "inputs": [{
                "referenceName": "SourceDataset",
                "type": "DatasetReference"
            }],
            "outputs": [{
                "referenceName": "SinkDataset",
                "type": "DatasetReference"
            }]
        }]
    }
}

Set-AzDataFactoryV2Pipeline コマンドレットを実行して、パイプラインFullCopyPipeline を作成します。

 Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "FullCopyPipeline" -File ".\FullCopyPipeline.json"

出力例を次に示します。

 PipelineName      : FullCopyPipeline
 ResourceGroupName : ADFTutorialResourceGroup
 DataFactoryName   : IncCopyChgTrackingDF
 Activities        : {FullCopyActivity}
 Parameters        :

フルコピーパイプラインを実行する

パイプライン FullCopyPipeline を、Invoke-AzDataFactoryV2Pipeline コマンドレットを使って実行します。

Invoke-AzDataFactoryV2Pipeline -PipelineName "FullCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName

フルコピーパイプラインを監視する

Azure portal にログインします。
[すべてのサービス] をクリックし、キーワード「data factories」で検索して、 [データファクトリ] を選択します。
データファクトリの一覧から目的のデータファクトリを探して選択し、[データファクトリ] ページを起動します。
[データファクトリ] ページの [監視と管理] タイルをクリックします。
独立したタブでデータ統合アプリケーションが起動します。すべてのパイプラインの実行とその状態を確認できます。次の例では、パイプラインの実行が、成功状態であることに注目してください。 [パラメーター] 列のリンクをクリックすると、パイプラインに渡されたパラメーターを確認できます。エラーが発生した場合は、 [エラー] 列にリンクが表示されます。 [アクション] 列のリンクをクリックします。
[アクション] 列のリンクをクリックすると、以下のページが開いて、そのパイプラインに関するすべてのアクティビティの実行が表示されます。
再度パイプラインの実行ビューに切り替えるには、この画像に示した [パイプライン] をクリックします。

結果の確認

incremental-<GUID>.txt コンテナーの incchgtracking フォルダーに、adftutorial という名前のファイルが確認できます。

フルコピーからの出力ファイル

このファイルには、データベースからのデータが存在します。

1,aaaa,21
2,bbbb,24
3,cccc,20
4,dddd,26
5,eeee,22

ソーステーブルにデータを追加する

データベースに次のクエリを実行して行の追加と更新を行います。

INSERT INTO data_source_table
(PersonID, Name, Age)
VALUES
(6, 'new','50');


UPDATE data_source_table
SET [Age] = '10', [name]='update' where [PersonID] = 1

差分コピーのパイプラインを作成する

この手順では、次のアクティビティを含んだパイプラインを作成して定期的に実行します。 lookup アクティビティAzure SQL Databaseから新旧のSYS_CHANGE_VERSIONを取得し、コピーアクティビティに渡します。 copy アクティビティは、2 つのSYS_CHANGE_VERSION値の間に挿入/更新/削除されたデータをAzure SQL DatabaseからAzure Blob Storageにコピーします。 ストアドプロシージャアクティビティは、次回のパイプライン実行に備えて SYS_CHANGE_VERSION の値を更新します。

JSON ファイル IncrementalCopyPipeline.json を同じフォルダー内に作成します。次の内容を記述してください。

{
    "name": "IncrementalCopyPipeline",
    "properties": {
        "activities": [
            {
                "name": "LookupLastChangeTrackingVersionActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select * from table_store_ChangeTracking_version"
                    },
                    "dataset": {
                        "referenceName": "ChangeTrackingDataset",
                        "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "LookupCurrentChangeTrackingVersionActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion"
                    },
                    "dataset": {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "IncrementalCopyActivity",
                "type": "Copy",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select data_source_table.PersonID,data_source_table.Name,data_source_table.Age, CT.SYS_CHANGE_VERSION, SYS_CHANGE_OPERATION from data_source_table RIGHT OUTER JOIN CHANGETABLE(CHANGES data_source_table, @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.SYS_CHANGE_VERSION}) as CT on data_source_table.PersonID = CT.PersonID where CT.SYS_CHANGE_VERSION <= @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}"
                    },
                    "sink": {
                        "type": "BlobSink"
                    }
                },
                "dependsOn": [
                    {
                        "activity": "LookupLastChangeTrackingVersionActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    },
                    {
                        "activity": "LookupCurrentChangeTrackingVersionActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "inputs": [
                    {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference"
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "SinkDataset",
                        "type": "DatasetReference"
                    }
                ]
            },
            {
                "name": "StoredProceduretoUpdateChangeTrackingActivity",
                "type": "SqlServerStoredProcedure",
                "typeProperties": {
                    "storedProcedureName": "Update_ChangeTracking_Version",
                    "storedProcedureParameters": {
                        "CurrentTrackingVersion": {
                            "value": "@{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}",
                            "type": "INT64"
                        },
                        "TableName": {
                            "value": "@{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}",
                            "type": "String"
                        }
                    }
                },
                "linkedServiceName": {
                    "referenceName": "AzureSQLDatabaseLinkedService",
                    "type": "LinkedServiceReference"
                },
                "dependsOn": [
                    {
                        "activity": "IncrementalCopyActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ]
            }
        ]
    }
}

Set-AzDataFactoryV2Pipeline コマンドレットを実行して、パイプラインFullCopyPipeline を作成します。

 Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "IncrementalCopyPipeline" -File ".\IncrementalCopyPipeline.json"

出力例を次に示します。

 PipelineName      : IncrementalCopyPipeline
 ResourceGroupName : ADFTutorialResourceGroup
 DataFactoryName   : IncCopyChgTrackingDF
 Activities        : {LookupLastChangeTrackingVersionActivity, LookupCurrentChangeTrackingVersionActivity, IncrementalCopyActivity, StoredProceduretoUpdateChangeTrackingActivity}
 Parameters        :

増分コピーパイプラインを実行する

パイプライン IncrementalCopyPipeline を、Invoke-AzDataFactoryV2Pipeline コマンドレットを使って実行します。

Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroup $resourceGroupName -dataFactoryName $dataFactoryName

インクリメンタルコピーパイプラインを監視する

データ統合アプリケーションで、パイプラインの実行ビューを最新の情報に更新します。その一覧に IncrementalCopyPipeline があることを確認してください。 [アクション] 列のリンクをクリックします。

スクリーンショットには、あなたのパイプラインを含むデータファクトリのパイプライン実行が示されています。
[アクション] 列のリンクをクリックすると、以下のページが開いて、そのパイプラインに関するすべてのアクティビティの実行が表示されます。
再度パイプラインの実行ビューに切り替えるには、この画像に示した [パイプライン] をクリックします。

結果の確認

incchgtracking コンテナーの adftutorial フォルダーにもう 1 つファイルが確認できます。

増分コピーからの出力ファイル

このファイルに含まれているのは、データベースからの差分データのみです。 U と記録されているレコードはデータベース内の更新された行で、I は追加された行です。

1,update,10,2,U
6,new,50,1,I

最初の 3 つの列は、data_source_table 内の変更済みデータです。最後の 2 つの列は、変更追跡システムテーブルからのメタデータです。 4 つ目の列は、変更済みの各行の SYS_CHANGE_VERSION です。 5 つ目の列は、実行された操作で、U は更新 (Update) を、I は挿入 (Insert) を表します。変更追跡情報の詳細については、「CHANGETABLE」を参照してください。

==================================================================
PersonID Name    Age    SYS_CHANGE_VERSION    SYS_CHANGE_OPERATION
==================================================================
1        update  10            2                                 U
6        new     50            1                                 I

次のチュートリアルに進んで、LastModifiedDate に基づいて新規ファイルおよび変更されたファイルのみをコピーする方法について学習してください。

lastmodifieddate を基準に新しいファイルをコピーする

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-04-07

次の方法で共有

PowerShell を使用して変更追跡情報を使用してAzure SQL DatabaseからAzure Blob Storageにデータを増分読み込みする

概要

エンド ツー エンド ワークフロー

高レベルのソリューション

前提条件

データベースにデータ ソース テーブルを作成する

Azure PowerShell

Data Factory の作成

リンクされたサービスを作成します

Azure Storage リンク サービスを作成します。

Azure SQL Database のリンク先サービスを作成します。

データセットを作成する

ソース データセットを作成する

シンク データセットを作成する

変更追跡データセットを作成する

完全コピーのパイプラインを作成する

フル コピー パイプラインを実行する

フル コピー パイプラインを監視する

結果の確認

ソース テーブルにデータを追加する

差分コピーのパイプラインを作成する

増分コピー パイプラインを実行する

インクリメンタルコピー パイプラインを監視する

結果の確認

関連するコンテンツ

フィードバック

その他のリソース

エンドツーエンドワークフロー

データベースにデータソーステーブルを作成する

Azure Storage リンクサービスを作成します。

ソースデータセットを作成する

シンクデータセットを作成する

フルコピーパイプラインを実行する

フルコピーパイプラインを監視する

ソーステーブルにデータを追加する

増分コピーパイプラインを実行する

インクリメンタルコピーパイプラインを監視する