Share via

Memuat data secara bertambah bertahap dari Azure SQL Database ke Azure Blob Storage dengan informasi pelacakan perubahan menggunakan PowerShell

  • Artikel

BERLAKU UNTUK:Azure Data Factory Azure Synapse Analytics

Tip

Cobalah Data Factory di Microsoft Fabric, solusi analitik all-in-one untuk perusahaan. Microsoft Fabric mencakup semuanya mulai dari pergerakan data hingga ilmu data, analitik real time, kecerdasan bisnis, dan pelaporan. Pelajari cara memulai uji coba baru secara gratis!

Dalam tutorial ini, Anda akan membuat pabrik data Azure dengan alur yang memuat data delta berdasarkan informasi pelacakan perubahan di database sumber di Azure SQL Database ke penyimpanan blob Azure.

Anda akan melakukan langkah-langkah berikut dalam tutorial ini:

  • Menyiapkan penyimpanan data sumber
  • Membuat pabrik data.
  • Membuat layanan tertaut.
  • Membuat himpunan data sumber, sink, dan pelacakan perubahan.
  • Membuat, menjalankan, dan memantau alur salinan lengkap
  • Menambahkan atau memperbarui data dalam tabel sumber
  • Membuat, menjalankan, dan memantau alur salinan yang bertambah bertahap

Catatan

Sebaiknya Anda menggunakan modul Azure Az PowerShell untuk berinteraksi dengan Azure. Lihat Menginstal Azure PowerShell untuk memulai. Untuk mempelajari cara bermigrasi ke modul Az PowerShell, lihat Memigrasikan Azure PowerShell dari AzureRM ke Az.

Gambaran Umum

Dalam solusi integrasi data, memuat data secara bertambah bertahap setelah muatan data awal adalah skenario yang banyak digunakan. Dalam beberapa kasus, data yang diubah dalam suatu periode di penyimpanan data sumber Anda dapat dengan mudah dipotong (misalnya, LastModifyTime, CreationTime). Dalam beberapa kasus, tidak ada cara eksplisit untuk mengidentifikasi data delta dari kali terakhir Anda memproses data. Teknologi Pelacakan Perubahan yang didukung oleh penyimpanan data seperti Azure SQL Database dan SQL Server dapat digunakan untuk mengidentifikasi data delta. Tutorial ini menjelaskan cara menggunakan Azure Data Factory dengan teknologi Pelacakan Perubahan SQL untuk memuat data delta secara bertambah bertahap dari Azure SQL Database ke Azure Blob Storage. Untuk informasi yang lebih konkret tentang teknologi Pelacakan Perubahan SQL, lihat Pelacakan perubahan di SQL Server.

Alur kerja end-to-end

Berikut adalah langkah alur kerja ujung-ke-ujung umum untuk memuat data secara bertambah bertahap menggunakan teknologi Pelacakan Perubahan.

Catatan

Baik Azure SQL Database maupun SQL Server mendukung teknologi Pelacakan Perubahan. Tutorial ini menggunakan Azure SQL Database sebagai penyimpanan data sumber. Anda juga dapat menggunakan instans SQL Server.

  1. Pemuatan awal data historis (jalankan sekali):
    1. Aktifkan teknologi Pelacakan Perubahan dalam database sumber di Azure SQL Database.
    2. Dapatkan nilai awal SYS_CHANGE_VERSION di database sebagai garis besar untuk mengambil data yang diubah.
    3. Muat data lengkap dari database sumber ke penyimpanan blob Azure.
  2. Pemuatan data delta inkremental pada jadwal (berjalan secara berkala setelah pemuatan awal data):
    1. Dapatkan nilai SYS_CHANGE_VERSION lama dan baru.
    2. Muat data delta dengan menggabungkan kunci utama baris yang diubah (di antara dua nilai SYS_CHANGE_VERSION) dari sys.change_tracking_tables dengan data dalam tabel sumber, lalu pindahkan data delta ke tujuan.
    3. Perbarui SYS_CHANGE_VERSION untuk pemuatan delta selanjutnya.

Solusi tingkat tinggi

Dalam tutorial ini, Anda akan membuat dua alur yang melakukan dua operasi berikut:

  1. Muatan awal: Anda akan membuat alur dengan aktivitas salin yang menyalin seluruh data dari penyimpanan data sumber (Azure SQL Database) ke penyimpanan data tujuan (Azure Blob Storage).

    Full loading of data

  2. Muatan bertambah bertahap: Anda akan membuat alur dengan aktivitas berikut, dan menjalankannya secara berkala.

    1. Buat dua aktivitas pencarian untuk mendapatkan file SYS_CHANGE_VERSION lama dan baru dari Azure SQL Database lalu meneruskannya ke aktivitas salin.
    2. Buat satu aktivitas salin untuk menyalin data yang disisipkan/diperbarui/dihapus antara dua nilai SYS_CHANGE_VERSION dari Azure SQL Database ke Azure Blob Storage.
    3. Buat satu aktivitas prosedur tersimpan untuk memperbarui nilai SYS_CHANGE_VERSION untuk eksekusi alur berikutnya.

    Increment load flow diagram

Jika Anda tidak memiliki langganan Azure, buat akun gratis sebelum Anda memulai.

Prasyarat

  • Azure PowerShell. Pasang modul Azure PowerShell terbaru dengan mengikuti petunjuk dalam Cara menginstal dan mengonfigurasi Azure PowerShell.
  • Microsoft Azure SQL database. Anda menggunakan database sebagai penyimpanan datasumber. Jika Anda tidak memiliki database di Azure SQL Database, lihat artikel Buat database di Azure SQL Database untuk mengetahui langkah-langkah membuatnya.
  • Akun Azure Storage. Anda menggunakan blob penyimpanan sebagai sink penyimpanan data. Jika Anda tidak memiliki akun penyimpanan Azure, lihat artikel Buat akun penyimpanan untuk mengetahui langkah-langkah membuatnya. Buat kontainer dengan nama adftutorial.

Membuat tabel sumber data di database Anda

  1. Buka SQL Server Management Studio, lalu hubungkan ke SQL Database.

  2. Di Server Explorer, klik kanan database lalu pilih Kueri Baru.

  3. Jalankan perintah SQL berikut terhadap database untuk membuat tabel bernama data_source_table sebagai penyimpanan sumber data.

    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);

  4. Aktifkan mekanisme Pelacakan Perubahan di database Anda dan tabel sumber (data_source_table) dengan menjalankan kueri SQL berikut:

    Catatan

    • Ganti <nama database> dengan nama database Anda yang memiliki data_source_table.
    • Data yang diubah telah disimpan selama dua hari di contoh saat ini. Jika Anda memuat data yang diubah setiap tiga hari atau lebih, beberapa data yang diubah tidak akan disertakan. Anda perlu mengubah nilai CHANGE_RETENTION ke angka yang lebih besar. Atau, pastikan bahwa periode Anda untuk memuat data yang diubah adalah dalam waktu dua hari. Untuk informasi selengkapnya, lihat Mengaktifkan pelacakan perubahan untuk database
    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)

  5. Buat tabel baru dan simpan ChangeTracking_version dengan nilai default dengan menjalankan kueri berikut:

    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)

    Catatan

    Jika data tidak berubah setelah Anda mengaktifkan pelacakan perubahan untuk SQL Database, nilai versi pelacakan perubahan tersebut adalah 0.

  6. Jalankan perintah berikut untuk membuat prosedur tersimpan di database Anda. Alur memanggil prosedur tersimpan ini untuk memperbarui versi pelacakan perubahan di tabel yang Anda buat di langkah sebelumnya.

    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

Pasang modul Azure PowerShell terbaru dengan mengikuti petunjuk dalam Cara menginstal dan mengonfigurasi Azure PowerShell.

Membuat pabrik data

  1. Tentukan variabel untuk nama grup sumber daya yang Anda gunakan di perintah PowerShell nanti. Salin teks perintah berikut ke PowerShell, tentukan nama untuk grup sumber daya Azure dalam tanda kutip ganda, lalu jalankan perintah. Sebagai contoh: "adfrg".

    $resourceGroupName = "ADFTutorialResourceGroup";

    Jika grup sumber daya sudah ada, Anda mungkin tidak ingin menimpanya. Tetapkan nilai yang berbeda ke variabel $resourceGroupName dan jalankan perintah lagi

  2. Tentukan variabel untuk lokasi pabrik data:

    $location = "East US"

  3. Untuk membuat grup sumber daya Azure, jalankan perintah berikut:

    New-AzResourceGroup $resourceGroupName $location

    Jika grup sumber daya sudah ada, Anda mungkin tidak ingin menimpanya. Tetapkan nilai yang berbeda ke variabel $resourceGroupName dan jalankan perintah lagi.

  4. Tentukan variabel untuk nama pabrik data.

    Penting

    Perbarui nama pabrik data menjadi nama yang unik secara global.

    $dataFactoryName = "IncCopyChgTrackingDF";

  5. Untuk membuat pabrik data, jalankan cmdlet Set-AzDataFactoryV2 berikut:

    Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location $location -Name $dataFactoryName

Perhatikan poin berikut:

  • Nama pabrik data Azure harus bersifat unik secara global. Jika Anda menerima kesalahan berikut, ubah nama dan coba lagi.

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

  • Untuk membuat instans Data Factory, akun pengguna yang Anda gunakan untuk masuk ke Azure harus merupakan anggota dari peran kontributor atau pemilik, atau administrator dari langganan Azure.

  • Untuk daftar wilayah Azure tempat Data Factory saat ini tersedia, pilih wilayah yang menarik minat Anda pada halaman berikut, lalu perluas Analitik untuk menemukan Data Factory: Produk yang tersedia menurut wilayah. Penyimpanan data (Azure Storage, Azure SQL Database, dll.) dan komputasi (HDInsight, dll.) yang digunakan oleh pabrik data dapat berada di wilayah lain.

Membuat layanan tertaut

Anda membuat layanan tertaut di pabrik data untuk menautkan penyimpanan data dan layanan komputasi ke pabrik data. Di bagian ini, Anda akan membuat layanan tertaut ke akun Azure Storage dan database Anda di Azure SQL Database.

Membuat layanan tertaut Azure Storage.

Di langkah ini, Anda akan menautkan Akun Azure Storage ke pabrik data.

  1. Buat file JSON bernama AzureStorageLinkedService.json di folder C:\ADFTutorials\IncCopyChangeTrackingTutorial dengan konten berikut: (Buat folder jika belum ada.). Ganti <accountName>, <accountKey> dengan nama dan kunci akun penyimpanan Azure Anda sebelum menyimpan file.

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

  2. Di Azure PowerShell, beralih ke folder C:\ADFTutorials\IncCopyChangeTrackingTutorial.

  3. Jalankan cmdlet Set-AzDataFactoryV2LinkedService untuk membuat layanan yang ditautkan: AzureStorageLinkedService. Dalam contoh berikut, Anda meneruskan nilai untuk parameter ResourceGroupName dan DataFactoryName.

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

    Berikut adalah output sampel:

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

Buat layanan tertaut Azure SQL Database.

Dalam langkah ini, Anda akan menautkan database ke pabrik data.

  1. Buat file JSON bernama AzureSQLDatabaseLinkedService.json di folder C:\ADFTutorials\IncCopyChangeTrackingTutorial dengan konten berikut: Ganti <server><nama database>, <id pengguna>, dan <kata sandi> dengan nama server Anda, nama database, ID pengguna, dan kata sandi Anda sebelum menyimpan file.

    {
    "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;"
        }
    }
}

  2. Di Azure PowerShell,jalankan cmdlet Set-AzDataFactoryV2LinkedService untuk membuat layanan tertaut: AzureSQLDatabaseLinkedService.

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

    Berikut adalah output sampel:

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

Membuat himpunan data

Dalam langkah ini, Anda akan membuat himpunan data untuk mewakili sumber data, tujuan data. dan tempat untuk menyimpan SYS_CHANGE_VERSION.

Membuat himpunan data sumber

Dalam langkah ini, Anda akan membuat himpunan data untuk mewakili data sumber.

  1. Buat file JSON bernama SourceDataset.json di folder yang sama dengan konten berikut:

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

  2. Jalankan cmdlet Set-AzDataFactoryV2Dataset untuk membuat himpunan data: SourceDataset

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

    Berikut adalah output sampel dari cmdlet tersebut:

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

Membuat himpunan data sink

Dalam langkah ini, Anda akan membuat himpunan data untuk mewakili data yang disalin dari penyimpanan data sumber.

  1. Buat file JSON bernama SinkDataset.json di folder yang sama dengan konten berikut:

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

    Anda akan membuat kontainer adftutorial di Azure Blob Storage sebagai bagian dari prasyarat. Buat kontainer jika belum ada (atau) atur ke nama kontainer yang sudah ada. Dalam tutorial ini, nama file output dihasilkan secara dinamis dengan menggunakan ekspresi: @CONCAT('Incremental-', alur().RunId, '.txt').

  2. Jalankan cmdlet Set-AzDataFactoryV2Dataset untuk membuat himpunan data: SinkDataset

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

    Berikut adalah output sampel dari cmdlet tersebut:

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

Membuat himpunan data pelacakan perubahan

Dalam langkah ini, Anda akan membuat himpunan data untuk menyimpan versi pelacakan perubahan.

  1. Buat file JSON bernama ChangeTrackingDataset.json di folder yang sama dengan konten berikut:

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

    Anda membuat tabel table_store_ChangeTracking_version sebagai bagian dari prasyarat.

  2. Jalankan cmdlet Set-AzDataFactoryV2Dataset untuk membuat himpunan data: ChangeTrackingDataset

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

    Berikut adalah output sampel dari cmdlet tersebut:

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

Membuat alur untuk salinan lengkap

Dalam langkah ini, Anda akan membuat alur dengan aktivitas salin yang menyalin seluruh data dari penyimpanan data sumber (Azure SQL Database) ke penyimpanan data tujuan (Azure Blob Storage).

  1. Buat file JSON: FullCopyPipeline.json di folder yang sama dengan konten berikut:

    {
    "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"
            }]
        }]
    }
}

  2. Jalankan cmdlet Set-AzDataFactoryV2Pipeline untuk membuat alur: FullCopyPipeline.

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

    Berikut adalah output sampel:

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

Menjalankan alur salinan lengkap

Jalankan alur: FullCopyPipeline dengan menggunakan cmdlet Invoke-AzDataFactoryV2Pipeline.

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

Memantau alur salinan lengkap

  1. Masuk ke portal Microsoft Azure.

  2. Klik Semua layanan, cari dengan kata kunci data factories, dan pilih Pabrik data.

    Data factories menu

  3. Cari pabrik data Anda di daftar pabrik data, dan pilih untuk meluncurkan halaman Pabrik data.

    Search for your data factory

  4. Di halaman Pabrik data, klik petak peta Pantau & Kelola.

    Monitor & Manage tile

  5. Aplikasi Integrasi Data diluncurkan di tab terpisah. Anda dapat melihat semua eksekusi alur dan statusnya. Perhatikan bahwa dalam contoh berikut, status eksekusi alur menjadi Berhasil. Anda dapat memeriksa parameter yang diteruskan ke alur dengan mengklik tautan di kolom Parameter. Jika ada kesalahan, Anda akan melihat tautan di kolom Kesalahan. Klik tautan di kolom Tindakan.

    Screenshot shows pipeline runs for a data factory.

  6. Saat mengklik tautan di kolom Tindakan, Anda akan melihat halaman berikut yang memperlihatkan semua eksekusi aktivitas untuk alur.

    Screenshot shows activity runs for a data factory with the Pipelines link called out.

  7. Untuk beralih kembali ke tampilan Eksekusi alur, klik Alur seperti yang ditunjukkan di gambar.

Meninjau hasil

Anda akan melihat file dengan nama incremental-<GUID>.txt dalam folder incchgtracking dari kontainer adftutorial.

Output file from full copy

File tersebut harus berisi data dari database Anda:

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

Menambahkan lebih banyak data ke tabel sumber

Jalankan kueri berikut ini terhadap database Anda untuk menambahkan baris dan memperbarui baris.

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


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

Membuat alur untuk salinan delta

Dalam langkah ini, Anda akan membuat alur dengan aktivitas berikut, dan menjalankannya secara berkala. Aktivitas pencarian akan mendapatkan file SYS_CHANGE_VERSION lama dan baru dari Azure SQL Database lalu meneruskannya ke aktivitas salin. Aktivitas salin akan menyalin data yang disisipkan/diperbarui/dihapus antara dua nilai SYS_CHANGE_VERSION dari Azure SQL Database ke Azure Blob Storage. Aktivitas prosedur tersimpan akan memperbarui nilai SYS_CHANGE_VERSION untuk eksekusi alur berikutnya.

  1. Buat file JSON: IncrementalCopyPipeline.json di folder yang sama dengan konten berikut:

    {
    "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"
                        ]
                    }
                ]
            }
        ]
    }
}

  2. Jalankan cmdlet Set-AzDataFactoryV2Pipeline untuk membuat alur: FullCopyPipeline.

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

    Berikut adalah output sampel:

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

Menjalankan alur salinan yang bertambah bertahap

Jalankan alur: IncrementalCopyPipeline dengan menggunakan cmdlet Invoke-AzDataFactoryV2Pipeline.

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

Memantau alur salin yang bertambah bertahap

  1. Dalam Aplikasi Integrasi Data, refresh tampilan eksekusi alur. Pastikan Anda melihat IncrementalCopyPipeline dalam daftar. Klik tautan di kolom Tindakan.

    Screenshot shows pipeline runs for a data factory including your pipeline.

  2. Saat mengklik tautan di kolom Tindakan, Anda akan melihat halaman berikut yang memperlihatkan semua eksekusi aktivitas untuk alur.

    Screenshot shows pipeline runs for a data factory with several marked as succeeded.

  3. Untuk beralih kembali ke tampilan Eksekusi alur, klik Alur seperti yang ditunjukkan di gambar.

Meninjau hasil

Anda akan melihat file kedua di folder incchgtrackingdari kontainer adftutorial.

Output file from incremental copy

File seharusnya hanya memiliki data delta dari database Anda saja. Rekaman dengan U adalah baris yang diperbarui dalam database dan I merupakan satu baris yang ditambahkan.

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

Tiga kolom pertama adalah data yang diubah dari data_source_table. Dua kolom terakhir adalah metadata dari tabel sistem pelacakan perubahan. Kolom keempat adalah SYS_CHANGE_VERSION setiap baris yang diubah. Kolom kelima adalah operasi: U = update, I = insert. Untuk detail tentang informasi pelacakan perubahan, lihat CHANGETABLE.

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

Konten terkait

Lanjutkan ke tutorial berikut untuk mempelajari cara tentang menyalin file baru dan yang diubah hanya berdasarkan LastModifiedDate file tersebut:

Menyalin file baru berdasarkan lastmodifieddate