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

[アーティクル]
10/20/2023

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータサイエンス、リアルタイム分析、ビジネスインテリジェンス、レポートまで、あらゆるものをカバーしています。無料で新しい試用版を開始する方法について説明します。

このチュートリアルでは、Azure SQL Database のソースデータベースから変更追跡情報に基づく差分データを Azure Blob Storage に読み込むパイプラインを使用して Azure Data Factory を作成します。

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

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

注意

Azure を操作するには、Azure Az PowerShell モジュールを使用することをお勧めします。作業を開始するには、Azure PowerShell のインストールに関する記事を参照してください。 Az PowerShell モジュールに移行する方法については、「AzureRM から Az への Azure PowerShell の移行」を参照してください。

概要

データ統合ソリューションでは、初回データ読み込みの後、増分データを読み込む手法が広く利用されています。ソースデータストアから特定の期間に変更されたデータを簡単に切り出すことができる場合もあります (LastModifyTime、CreationTime など)。一方、前回データを処理した時点からの差分データを明示的に特定する方法がない場合もあります。こうした差分データは、Azure SQL Database や SQL Server などのデータストアでサポートされる Change Tracking テクノロジを使用して特定することができます。このチュートリアルでは、Azure Data Factory と SQL Change Tracking テクノロジを使用して、Azure SQL Database から Azure Blob Storage に差分データを増分読み込みする方法について説明します。 SQL Change Tracking テクノロジに関するより具体的な情報については、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 Storage にフルデータを読み込みます。
スケジュール に基づく差分データの増分読み込み (データの初期読み込み後に定期的に実行):
1. 新旧の SYS_CHANGE_VERSION 値を取得します。
2. 差分データを読み込みます。これは、sys.change_tracking_tables から得られる変更済みの行 (2 つの SYS_CHANGE_VERSION 値の間にある行) の主キーとソーステーブル内のデータとを結合した後、その差分データをターゲットに移動することによって行います。
3. 次回の差分読み込みに備えて SYS_CHANGE_VERSION を更新します。

ソリューションの概略

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

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

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

前提条件

Azure PowerShell。 Azure PowerShell のインストールと構成の方法に関するページの手順に従って、最新の Azure PowerShell モジュールをインストールしてください。
Azure SQL データベース。 ソース データストアとして使うデータベースです。 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 のインストールと構成の方法に関するページの手順に従って、最新の 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 Data Factory の名前はグローバルに一意にする必要があります。次のエラーが発生した場合は、名前を変更してからもう一度実行してください。
```
The specified Data Factory name 'ADFIncCopyChangeTrackingTestFactory' is already in use. Data Factory names must be globally unique.
```
Data Factory インスタンスを作成するには、Azure へのログインに使用するユーザーアカウントが、共同作成者ロールまたは所有者ロールのメンバーであるか、Azure サブスクリプションの管理者である必要があります。
現在 Data Factory が利用できる Azure リージョンの一覧については、次のページで目的のリージョンを選択し、 [分析] を展開して [Data Factory] を探してください。リージョン別の利用可能な製品データファクトリで使用するデータストア (Azure Storage、Azure SQL Database など) やコンピューティング (HDInsight など) は他のリージョンに配置できます。

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

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

Azure Storage のリンクされたサービスを作成する

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

以下の内容を記述した 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 ファイルを作成します。<server><database name>、<user id>、<password> を実際のサーバーの名前、データベース名、ユーザー ID、パスワードに置き換えてからファイルを保存してください。

{
    "name": "AzureSQLDatabaseLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Server = tcp:<server>.database.windows.net,1433;Initial Catalog=<database name>; Persist Security Info=False; User ID=<user name>; Password=<password>; MultipleActiveResultSets = False; Encrypt = True; TrustServerCertificate = False; Connection Timeout = 30;"
        }
    }
}

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"
        }
    }
}
```
前提条件の 1 つとして adftutorial コンテナーを Azure Blob Storage に作成します。このコンテナーが存在しない場合は作成するか、または既存のコンテナーの名前に設定してください。このチュートリアルでは、@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」で検索して、 [データファクトリ] を選択します。
データファクトリの一覧から目的のデータファクトリを探して選択し、[データファクトリ] ページを起動します。
[データファクトリ] ページの [監視と管理] タイルをクリックします。
独立したタブでデータ統合アプリケーションが起動します。すべてのパイプラインの実行とその状態を確認できます。次の例では、パイプラインの実行が、成功状態であることに注目してください。 [パラメーター] 列のリンクをクリックすると、パイプラインに渡されたパラメーターを確認できます。エラーが発生した場合は、 [エラー] 列にリンクが表示されます。 [アクション] 列のリンクをクリックします。
[アクション] 列のリンクをクリックすると、以下のページが開いて、そのパイプラインに関するすべてのアクティビティの実行が表示されます。
再度パイプラインの実行ビューに切り替えるには、この画像に示した [パイプライン] をクリックします。

結果の確認

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

Output file from full copy

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

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

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

この手順では、次のアクティビティを含んだパイプラインを作成して定期的に実行します。 ルックアップアクティビティは、Azure SQL Database から新旧の SYS_CHANGE_VERSION を取得してコピーアクティビティに渡します。 コピーアクティビティは、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 があることを確認してください。 [アクション] 列のリンクをクリックします。
[アクション] 列のリンクをクリックすると、以下のページが開いて、そのパイプラインに関するすべてのアクティビティの実行が表示されます。
再度パイプラインの実行ビューに切り替えるには、この画像に示した [パイプライン] をクリックします。

結果の確認

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

Output file from incremental copy

このファイルに含まれているのは、データベースからの差分データのみです。 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 を基準に新しいファイルをコピーする

Share via

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

概要

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

ソリューションの概略

前提条件

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

Azure PowerShell

Data Factory の作成

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

Azure Storage のリンクされたサービスを作成する

Azure SQL Database のリンクされたサービスを作成する

データセットを作成する

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

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

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

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

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

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

結果の確認

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

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

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

増分コピーパイプラインを監視する

結果の確認

フィードバック

フィードバック

その他のリソース

Share via

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

概要

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

ソリューションの概略

前提条件

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

Azure PowerShell

Data Factory の作成

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

Azure Storage のリンクされたサービスを作成する

Azure SQL Database のリンクされたサービスを作成する

データセットを作成する

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

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

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

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

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

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

結果の確認

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

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

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

増分コピー パイプラインを監視する

結果の確認

関連するコンテンツ

フィードバック

フィードバック

その他のリソース

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

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

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

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

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

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

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

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

増分コピーパイプラインを監視する