使用 PowerShell 以累加方式將資料從 Azure SQL Database 載入到 Azure Blob 儲存體

發行項
04/25/2024

適用於：Azure Data Factory Azure Synapse Analytics

提示

試用 Microsoft Fabric 中的 Data Factory，這是適用於企業的全方位分析解決方案。 Microsoft Fabric 涵蓋從資料移動到資料科學、即時分析、商業智慧和報告的所有項目。了解如何免費開始新的試用！

在此教學課程中，您會使用 Azure Data Factory 來建立管線，以將差異資料從 Azure SQL Database 中的資料表載入至 Azure Blob 儲存體。

您會在本教學課程中執行下列步驟：

準備資料存放區來儲存水位線值。
建立資料處理站。
建立連結的服務。
建立來源、接收及水位線資料集。
建立管線。
執行管線。
監視管道執行。

概觀

高階解決方案圖表如下：

Incrementally load data

以下是建立此解決方案的重要步驟：

選取水位線資料行。選取來源資料存放區中的一個資料行，可用於切割每次執行時新增或更新的記錄。一般來說，當建立或更新資料列時，這個選取的資料行 (例如，last_modify_time 或 ID) 中的資料會持續增加。此資料行中的最大值就作為水位線。
準備資料存放區來儲存水位線值。
在本教學課程中，您會將水位線值儲存在 SQL 資料庫中。
使用下列工作流程建立管線：

此解決方案中的管道有下列活動：
- 建立兩個查閱活動。使用第一個查閱活動來取出最後一個水位線值。使用第二個查閱活動來取出新的水位線值。這些水位線值會傳遞給複製活動。
- 建立複製活動，以複製來源資料存放區的資料列，而這些資料列的水位線資料行值大於舊水位線值，且小於或等於新水位線值。然後，它會將來源資料存放區的差異資料複製到 Blob 儲存體作為新檔案。
- 建立 StoredProcedure 活動，以更新下次執行的管線水位線值。

如果您沒有 Azure 訂用帳戶，請在開始前建立免費帳戶。

必要條件

注意

建議您使用 Azure Az PowerShell 模組來與 Azure 互動。請參閱安裝 Azure PowerShell 以開始使用。若要了解如何移轉至 Az PowerShell 模組，請參閱將 Azure PowerShell 從 AzureRM 移轉至 Az。

Azure SQL Database。您需要使用資料庫作為來源資料存放區。如果您在 Azure SQL Database 中沒有資料庫，請參閱在 Azure SQL Database 中建立資料庫，按照步驟建立資料庫。
Azure 儲存體。您需要使用 Blob 儲存體作為接收資料存放區。如果您沒有儲存體帳戶，請參閱建立儲存體帳戶，按照步驟來建立儲存體帳戶。建立名為 adftutorial 的容器。
Azure PowerShell。遵循安裝和設定 Azure PowerShell 中的指示。

在 SQL 資料庫中建立資料來源資料表

開啟 SQL Server Management Studio。在 [伺服器總管] 中，以滑鼠右鍵按一下資料庫，然後選擇 [新增查詢]。

對 SQL 資料庫執行下列 SQL 命令，以建立名為 data_source_table 的資料表作為資料來源存放區：

create table data_source_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

INSERT INTO data_source_table
(PersonID, Name, LastModifytime)
VALUES
(1, 'aaaa','9/1/2017 12:56:00 AM'),
(2, 'bbbb','9/2/2017 5:23:00 AM'),
(3, 'cccc','9/3/2017 2:36:00 AM'),
(4, 'dddd','9/4/2017 3:21:00 AM'),
(5, 'eeee','9/5/2017 8:06:00 AM');

在本教學課程中，您會使用 LastModifytime 作為水位線資料行。下表顯示資料來源存放區中的資料：

PersonID | Name | LastModifytime
-------- | ---- | --------------
1 | aaaa | 2017-09-01 00:56:00.000
2 | bbbb | 2017-09-02 05:23:00.000
3 | cccc | 2017-09-03 02:36:00.000
4 | dddd | 2017-09-04 03:21:00.000
5 | eeee | 2017-09-05 08:06:00.000

在 SQL 資料庫中建立另一個資料表來儲存高水位線值

對 SQL 資料庫執行下列 SQL 命令，以建立名為 watermarktable 的資料表來儲存水位線值：
```
create table watermarktable
(

TableName varchar(255),
WatermarkValue datetime,
);
```
使用來源資料存放區的資料表名稱來設定高水位線的預設值。在本教學課程中，資料表名稱是 data_source_table。
```
INSERT INTO watermarktable
VALUES ('data_source_table','1/1/2010 12:00:00 AM')    
```

檢閱資料表 watermarktable 中的資料。

Select * from watermarktable

輸出：

TableName  | WatermarkValue
----------  | --------------
data_source_table | 2010-01-01 00:00:00.000

在 SQL 資料庫中建立預存程序

執行下列命令，在您的 SQL 資料庫中建立預存程序：

CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS

BEGIN

UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName

END

建立資料處理站

定義資源群組名稱的變數，以便稍後在 PowerShell 命令中使用。將下列命令文字複製到 PowerShell，以雙引號指定 Azure 資源群組的名稱，然後執行命令。例如 "adfrg"。
```
$resourceGroupName = "ADFTutorialResourceGroup";
```
不建議您覆寫已經存在的資源群組。將不同的值指派給 $resourceGroupName 變數，然後再執行一次命令。
定義資料處理站位置的變數。
```
$location = "East US"
```
若要建立 Azure 資源群組，請執行下列命令：
```
New-AzResourceGroup $resourceGroupName $location
```
不建議您覆寫已經存在的資源群組。將不同的值指派給 $resourceGroupName 變數，然後再執行一次命令。
定義 Data Factory 名稱的變數。

重要

更新資料處理站名稱，使其成為全域唯一的資料處理站。例如，ADFTutorialFactorySP1127。
```
$dataFactoryName = "ADFIncCopyTutorialFactory";
```

若要建立資料處理站，請執行下列 Set-AzDataFactoryV2 Cmdlet：

Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location "East US" -Name $dataFactoryName

請注意下列幾點：

資料處理站的名稱必須是全域唯一的名稱。如果發生下列錯誤，請變更名稱，並再試一次：

The specified Data Factory name 'ADFv2QuickStartDataFactory' is already in use. Data Factory names must be globally unique.

若要建立 Data Factory 執行個體，您用來登入 Azure 的使用者帳戶必須為參與者或擁有者角色，或是 Azure 訂用帳戶的管理員。
如需目前可使用 Data Factory 的 Azure 區域清單，請在下列頁面上選取您感興趣的區域，然後展開 [分析] 以找出 [Data Factory]：依區域提供的產品。資料處理站所使用的資料存放區 (儲存體、SQL Database、Azure SQL 受控執行個體等) 和計算 (Azure HDInsight 等) 可位於其他區域。

建立連結服務

您在資料處理站中建立的連結服務會將您的資料存放區和計算服務連結到資料處理站。在本節中，您會對儲存體帳戶和 SQL Database 建立連結服務。

建立儲存體連結服務

使用下列內容，在 C:\ADF 資料夾中建立名為 AzureStorageLinkedService.json 的 JSON 檔案。 (建立資料夾 ADF (如果尚未存在)。)儲存檔案之前，以您的儲存體帳戶名稱和金鑰取代 <accountName> 和 <accountKey>。

{
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}

在 PowerShell 中，切換至 ADF 資料夾。

執行 Set-AzDataFactoryV2LinkedService Cmdlet 來建立連結服務 AzureStorageLinkedService。在下列範例中，您會傳遞 ResourceGroupName 和 DataFactoryName 參數的值：

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

以下是範例輸出:

LinkedServiceName : AzureStorageLinkedService
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureStorageLinkedService

建立 SQL Database 連結服務

使用下列內容，在 C:\ADF 資料夾中建立名為 AzureSQLDatabaseLinkedService.json 的 JSON 檔案。 (建立資料夾 ADF (如果尚未存在)。)儲存檔案之前，以您的伺服器名稱、資料庫、使用者識別碼和密碼，取代 <server>、<database>、<user id> 和 <password>。

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

在 PowerShell 中，切換至 ADF 資料夾。

執行 Set-AzDataFactoryV2LinkedService Cmdlet 來建立連結服務 AzureSQLDatabaseLinkedService。

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

以下是範例輸出:

LinkedServiceName : AzureSQLDatabaseLinkedService
ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService
ProvisioningState :

建立資料集

在此步驟中，您會建立資料集以代表來源和接收資料。

建立來源資料集

在相同的資料夾中，使用下列內容建立名為 SourceDataset.json 的 JSON 檔案：

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

在本教學課程中，您使用的資料表名稱是 data_source_table。如果您使用不同名稱的資料表，請取代此名稱。

執行 Set-AzDataFactoryV2Dataset Cmdlet 來建立資料集 SourceDataset。

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

以下是此 Cmdlet 的範例輸出：

DatasetName       : SourceDataset
ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

建立接收資料集

在相同的資料夾中，使用下列內容建立名為 SinkDataset.json 的 JSON 檔案：

{
    "name": "SinkDataset",
    "properties": {
        "type": "AzureBlob",
        "typeProperties": {
            "folderPath": "adftutorial/incrementalcopy",
            "fileName": "@CONCAT('Incremental-', pipeline().RunId, '.txt')",
            "format": {
                "type": "TextFormat"
            }
        },
        "linkedServiceName": {
            "referenceName": "AzureStorageLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

重要

此程式碼片段假設您在 Blob 儲存體中有一個名為 adftutorial 的 Blob 容器。建立容器 (若不存在)，或設為現有容器的名稱。如果容器中沒有輸出資料夾 incrementalcopy，則會自動建立。在本教學課程中，您會使用運算式 @CONCAT('Incremental-', pipeline().RunId, '.txt') 來動態產生檔案名稱。

執行 Set-AzDataFactoryV2Dataset Cmdlet 來建立資料集 SinkDataset。

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

以下是此 Cmdlet 的範例輸出：

DatasetName       : SinkDataset
ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureBlobDataset

建立水位線的資料集

在此步驟中，您會建立資料集來儲存高水位線值。

在相同的資料夾中，使用下列內容建立名為 WatermarkDataset.json 的 JSON 檔案：

{
    "name": " WatermarkDataset ",
    "properties": {
        "type": "AzureSqlTable",
        "typeProperties": {
            "tableName": "watermarktable"
        },
        "linkedServiceName": {
            "referenceName": "AzureSQLDatabaseLinkedService",
            "type": "LinkedServiceReference"
        }
    }
}

執行 Set-AzDataFactoryV2Dataset Cmdlet 來建立資料集 WatermarkDataset。

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

以下是此 Cmdlet 的範例輸出：

DatasetName       : WatermarkDataset
ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

建立新管線

在本教學課程中，您會建立具有兩個查閱活動、一個複製活動和一個 StoredProcedure 活動的管線，這些活動都在一個管線中鏈結。

在相同的資料夾中，使用下列內容建立 JSON 檔案 IncrementalCopyPipeline.json：

{
    "name": "IncrementalCopyPipeline",
    "properties": {
        "activities": [
            {
                "name": "LookupOldWaterMarkActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                    "type": "SqlSource",
                    "sqlReaderQuery": "select * from watermarktable"
                    },

                    "dataset": {
                    "referenceName": "WatermarkDataset",
                    "type": "DatasetReference"
                    }
                }
            },
            {
                "name": "LookupNewWaterMarkActivity",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select MAX(LastModifytime) as NewWatermarkvalue from data_source_table"
                    },

                    "dataset": {
                    "referenceName": "SourceDataset",
                    "type": "DatasetReference"
                    }
                }
            },

            {
                "name": "IncrementalCopyActivity",
                "type": "Copy",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'"
                    },
                    "sink": {
                        "type": "BlobSink"
                    }
                },
                "dependsOn": [
                    {
                        "activity": "LookupNewWaterMarkActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    },
                    {
                        "activity": "LookupOldWaterMarkActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],

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

            {
                "name": "StoredProceduretoWriteWatermarkActivity",
                "type": "SqlServerStoredProcedure",
                "typeProperties": {

                    "storedProcedureName": "usp_write_watermark",
                    "storedProcedureParameters": {
                        "LastModifiedtime": {"value": "@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}", "type": "datetime" },
                        "TableName":  { "value":"@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}", "type":"String"}
                    }
                },

                "linkedServiceName": {
                    "referenceName": "AzureSQLDatabaseLinkedService",
                    "type": "LinkedServiceReference"
                },

                "dependsOn": [
                    {
                        "activity": "IncrementalCopyActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ]
            }
        ]

    }
}

執行 Set-AzDataFactoryV2Pipeline Cmdlet 來建立管線 IncrementalCopyPipeline。

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

以下是範例輸出:

 PipelineName      : IncrementalCopyPipeline
 ResourceGroupName : ADF
 DataFactoryName   : incrementalloadingADF
 Activities        : {LookupOldWaterMarkActivity, LookupNewWaterMarkActivity, IncrementalCopyActivity, StoredProceduretoWriteWatermarkActivity}
 Parameters        :

執行管線

使用 Invoke-AzDataFactoryV2Pipeline Cmdlet 來執行管線 IncrementalCopyPipeline。以您自己的資源群組和資料處理站名稱取代預留位置。
```
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroupName $resourceGroupName -dataFactoryName $dataFactoryName
```

執行 Get-AzDataFactoryV2ActivityRun Cmdlet 來檢查管線的狀態，直到您看到所有活動成功執行為止。在 RunStartedAfter 和 RunStartedBefore 參數中，以您自己的適當時間取代預留位置。在本教學課程中，您會使用 -RunStartedAfter "2017/09/14" 和 -RunStartedBefore "2017/09/15"。

Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $RunId -RunStartedAfter "<start time>" -RunStartedBefore "<end time>"

以下是範例輸出:

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupNewWaterMarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {NewWatermarkvalue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:42:42 AM
ActivityRunEnd    : 9/14/2017 7:42:50 AM
DurationInMs      : 7777
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupOldWaterMarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {TableName, WatermarkValue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:42:42 AM
ActivityRunEnd    : 9/14/2017 7:43:07 AM
DurationInMs      : 25437
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : IncrementalCopyActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {source, sink}
Output            : {dataRead, dataWritten, rowsCopied, copyDuration...}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:43:10 AM
ActivityRunEnd    : 9/14/2017 7:43:29 AM
DurationInMs      : 19769
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : StoredProceduretoWriteWatermarkActivity
PipelineRunId     : d4bf3ce2-5d60-43f3-9318-923155f61037
PipelineName      : IncrementalCopyPipeline
Input             : {storedProcedureName, storedProcedureParameters}
Output            : {}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 7:43:32 AM
ActivityRunEnd    : 9/14/2017 7:43:47 AM
DurationInMs      : 14467
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

檢閱結果

在 Blob 儲存體 (接收存放區) 中，您會看到資料已複製到 SinkDataset 中定義的檔案。在目前的教學課程中，檔案名稱是 Incremental- d4bf3ce2-5d60-43f3-9318-923155f61037.txt。開啟檔案，您可以看到檔案中的記錄與 SQL 資料庫中的資料相同。
```
1,aaaa,2017-09-01 00:56:00.0000000
2,bbbb,2017-09-02 05:23:00.0000000
3,cccc,2017-09-03 02:36:00.0000000
4,dddd,2017-09-04 03:21:00.0000000
5,eeee,2017-09-05 08:06:00.0000000
```
檢查 watermarktable 中的最新值。您會看到水位線值已更新。
```
Select * from watermarktable
```
以下是範例輸出:

TableName WatermarkValue

data_source_table 2017-09-05 8:06:00.000

TableName	WatermarkValue
data_source_table	2017-09-05 8:06:00.000

將資料插入資料來源存放區以確認差異資料載入

將新資料插入 SQL 資料庫 (資料來源存放區)。

INSERT INTO data_source_table
VALUES (6, 'newdata','9/6/2017 2:23:00 AM')

INSERT INTO data_source_table
VALUES (7, 'newdata','9/7/2017 9:01:00 AM')

SQL 資料庫中更新的資料如下：

PersonID | Name | LastModifytime
-------- | ---- | --------------
1 | aaaa | 2017-09-01 00:56:00.000
2 | bbbb | 2017-09-02 05:23:00.000
3 | cccc | 2017-09-03 02:36:00.000
4 | dddd | 2017-09-04 03:21:00.000
5 | eeee | 2017-09-05 08:06:00.000
6 | newdata | 2017-09-06 02:23:00.000
7 | newdata | 2017-09-07 09:01:00.000

使用 Invoke-AzDataFactoryV2Pipeline Cmdlet，再次執行管線 IncrementalCopyPipeline。以您自己的資源群組和資料處理站名稱取代預留位置。
```
$RunId = Invoke-AzDataFactoryV2Pipeline -PipelineName "IncrementalCopyPipeline" -ResourceGroupName $resourceGroupName -dataFactoryName $dataFactoryName
```

Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $RunId -RunStartedAfter "<start time>" -RunStartedBefore "<end time>"

以下是範例輸出:

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupNewWaterMarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {NewWatermarkvalue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:52:26 AM
ActivityRunEnd    : 9/14/2017 8:52:58 AM
DurationInMs      : 31758
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : LookupOldWaterMarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, dataset}
Output            : {TableName, WatermarkValue}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:52:26 AM
ActivityRunEnd    : 9/14/2017 8:52:52 AM
DurationInMs      : 25497
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : IncrementalCopyActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {source, sink}
Output            : {dataRead, dataWritten, rowsCopied, copyDuration...}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:53:00 AM
ActivityRunEnd    : 9/14/2017 8:53:20 AM
DurationInMs      : 20194
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : ADF
DataFactoryName   : incrementalloadingADF
ActivityName      : StoredProceduretoWriteWatermarkActivity
PipelineRunId     : 2fc90ab8-d42c-4583-aa64-755dba9925d7
PipelineName      : IncrementalCopyPipeline
Input             : {storedProcedureName, storedProcedureParameters}
Output            : {}
LinkedServiceName :
ActivityRunStart  : 9/14/2017 8:53:23 AM
ActivityRunEnd    : 9/14/2017 8:53:41 AM
DurationInMs      : 18502
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

在 blob 儲存體中，您會看到已建立另一個檔案。在本教學課程中，新的檔案名稱是 Incremental-2fc90ab8-d42c-4583-aa64-755dba9925d7.txt。開啟該檔案，您會在其中看到兩列記錄。
檢查 watermarktable 中的最新值。您會看到水位線值再次更新。
```
Select * from watermarktable
```
範例輸出：

TableName WatermarkValue

data_source_table 2017-09-07 09:01:00.000

TableName	WatermarkValue
data_source_table	2017-09-07 09:01:00.000

在本教學課程中，您已執行下列步驟：

準備資料存放區來儲存水位線值。
建立資料處理站。
建立連結的服務。
建立來源、接收及水位線資料集。
建立管線。
執行管線。
監視管道執行。

在本教學課程中，管線已從 Azure SQL Database 中的單一資料表將資料複製到 Blob 儲存體。請前進到下列教學課程，了解如何將資料從 SQL Server 資料庫中的多個資料表複製到 SQL 資料庫。

以累加方式將 SQL Server 中多個資料表的資料載入到 Azure SQL Database

使用 PowerShell 以累加方式將資料從 Azure SQL Database 載入到 Azure Blob 儲存體

概觀

必要條件

在 SQL 資料庫中建立資料來源資料表

在 SQL 資料庫中建立另一個資料表來儲存高水位線值

在 SQL 資料庫中建立預存程序

建立資料處理站

建立連結服務

建立儲存體連結服務

建立 SQL Database 連結服務

建立資料集

建立來源資料集

建立接收資料集

建立水位線的資料集

建立新管線

執行管線

檢閱結果

將資料插入資料來源存放區以確認差異資料載入

相關內容

其他資源