Adatok szakaszos betöltése az Azure SQL Database-ből az Azure Blob Storage-ba változáskövetési információk és a PowerShell segítségével

Vonatkozik: Azure Data Factory Azure Synapse Analytics

Tipp

Data Factory a Microsoft Fabric a Azure Data Factory következő generációja, egyszerűbb architektúrával, beépített AI-vel és új funkciókkal. Ha még nem ismerkedik az adatintegrációval, kezdje a Fabric Data Factoryvel. A meglévő ADF-számítási feladatok frissíthetők Fabric használatával, hogy elérjék az adatkutatás, a valós idejű elemzés és a jelentéskészítés új képességeit.

• A Fabric ingyenes próbaverzió indítása.
• Az Azure Data Factory-ből a Data Factory-be a Microsoft Fabricben

Ebben az oktatóanyagban egy Azure data factoryt hoz létre egy folyamattal, amely a change tracking adatok alapján tölt be változásadatokat a forrásadatbázisban Azure SQL Database egy Azure blobtárolóba.

Az oktatóanyagban az alábbi lépéseket fogja végrehajtani:

• A forrásadattár előkészítése
• Adat-előállító létrehozása
• Társított szolgáltatások létrehozása.
• Forrás, fogadó és változáskövetési adatkészletek létrehozása.
• A teljes másolási folyamat létrehozása, futtatása és figyelése
• A forrástáblában szereplő adatok hozzáadása vagy frissítése
• A növekményes másolási folyamat létrehozása, futtatása és figyelése

Megjegyzés

Javasoljuk, hogy az Azure Az PowerShell-modult használja a Azure használatához. Első lépésként lásd: Install Azure PowerShell. Az Az PowerShell-modulra való migrálásról az Migrate Azure PowerShell az AzureRM-ből az Az című témakörben olvashat.

Áttekintés

Adatintegrációs megoldások esetében gyakran használt forgatókönyv az adatok növekményes betöltése egy kezdeti, teljes adatbetöltést követően. Bizonyos esetekben a forrásadattárban lévő adott időszakbeli módosított adatok egyszerűen felszeletelhetők (például: LastModifyTime, CreationTime). Bizonyos esetekben nincs egyértelmű módja a legutóbbi adatfeldolgozás során keletkezett delta adatok azonosításának. Az adattárak, például Azure SQL Database és SQL Server által támogatott változáskövetési technológia használható a változásadatok azonosítására. Ez az oktatóanyag bemutatja, hogyan használható az Azure Data Factory az SQL Change Tracking technológiával a delta adatok fokozatos betöltésére az Azure SQL Database-ből az Azure Blob Storage-ba. Az SQL Change Tracking technológiával kapcsolatos további információkért lásd a változáskövetés SQL Serverben részt.

Teljes körű munkafolyamat

Íme az adatok változáskövetési technológiával történő növekményes betöltési munkafolyamatának részletes lépései.

Megjegyzés

Mind Azure SQL Database, mind SQL Server támogatja a Change Tracking technológiát. Ez az oktatóanyag Azure SQL Database használ forrásadattárként. Egy SQL Server-példányt is használhat.

1. Az előzményadatok kezdeti betöltése (egyszeri futtatás):
   1. A változáskövetési technológia engedélyezése a forrásadatbázisban Azure SQL Database.
   2. Kérje le a SYS_CHANGE_VERSION kezdeti értékét az adatbázisban a módosított adatok rögzítésének alapkonfigurációjaként.
   3. Töltsön be teljes adatokat a forrásadatbázisból egy Azure blobtárolóba.
2. A deltaadatok növekményes betöltése ütemezés szerint (az adatok kezdeti betöltése után rendszeres időközönként fut):
   1. Kérje le a SYS_CHANGE_VERSION régi és új értékét.
   2. A változásadatok betöltéséhez csatolja a sys.change_tracking_tables táblázat módosított sorainak (a két SYS_CHANGE_VERSION érték közötti) elsődleges kulcsait a forrástábla adataihoz, és azután mozgassa át a delta adatokat a céltáblába.
   3. A következő alkalommal frissítse a SYS_CHANGE_VERSION értékét a delta betöltésekor.

Magas szintű megoldás

Ebben az oktatóanyagban két csővezetéket hoz létre, amelyek az alábbi két műveletet hajtják végre:

1. Betöltés: olyan másolási tevékenységgel rendelkező folyamatot hoz létre, amely a teljes adatokat a forrásadattárból (Azure SQL Database) a céladattárba (Azure Blob Storage) másolja.

   

2. Növekményes betöltés: egy csővezetéket hoz létre a következő tevékenységekkel, és azt időszakosan futtatja.

   1. Hozzon létre két keresési tevékenységet a régi és az új SYS_CHANGE_VERSION lekéréséhez az Azure SQL Database-ből; és adja át a másolási tevékenységhez.
   2. Hozzon létre egy másolási tevékenységet a beszúrt/frissített/törölt adatok másolásához a két SYS_CHANGE_VERSION érték között az Azure SQL Database és az Azure Blob Storage között.
   3. Hozzon létre egy tárolt eljárási tevékenységet a SYS_CHANGE_VERSION értékének a következő folyamatfuttatáshoz való frissítéséhez.

   

Ha nem rendelkezik Azure-előfizetéssel, a kezdés előtt hozzon létre egy free fiókot.

Előfeltételek

• Azure PowerShell. Telepítse a legújabb Azure PowerShell modulokat a Hogyan telepítse és konfigurálja az Azure PowerShellt útmutató utasításaiban.
• Azure SQL Database. Ezt az adatbázist használjuk forrásadattárként. Ha nem rendelkezik adatbázissal Azure SQL Database, a létrehozás lépéseit a Adatbázis létrehozása Azure SQL Database cikkben találja.
• Azure Storage fiók. A blobtárolót használjuk cél adattárként. Ha nincs Azure tárfiókja, a létrehozás lépéseit a Tárfiók létrehozása cikkben találja. Hozzon létre egy tárolót adftutorial néven.

Adatforrástábla létrehozása az adatbázisban

1. Indítsa el SQL Server Management Studio, és csatlakozzon az SQL Database-hez.

2. A Kiszolgálókezelőben kattintson a jobb gombbal az adatbázisra, és válassza az Új lekérdezés elemet.

3. Futtassa a következő SQL-parancsot az adatbázison egy adatforrástárként elnevezett data_source_table tábla létrehozásához.

   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. Engedélyezze a változáskövetési mechanizmust az adatbázisban és a forrástáblában (data_source_table) az alábbi SQL-lekérdezés futtatásával:

   Megjegyzés

   • Cserélje le <az adatbázis nevét> a data_source_table tartalmazó adatbázis nevére.
   • Ebben a példában a rendszer két napig tárolja a módosított adatokat. Ha a módosított adatokat három naponta vagy annál ritkábban tölti be, nem minden módosított adat jelenik meg. Vagy növelje a CHANGE_RETENTION változó értékét egy nagyobb számra. vagy ügyeljen arra, hogy a módosított adatok betöltései között ne teljen el két napnál több idő. További információk: Változáskövetés engedélyezése adatbázishoz.

   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. Hozzon létre egy új táblát, és az alábbi lekérdezés futtatásával tárolja alapértelmezett értékkel a ChangeTracking_version változót:

   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)
   

   Megjegyzés

   Ha az adatok nem módosultak, miután engedélyezte a változáskövetést az SQL Database-ben, a változáskövetés verziójának értéke 0.

6. Futtassa a következő lekérdezést egy tárolt eljárás létrehozásához az adatbázisban. A folyamat ezt a tárolt eljárást hívja meg az előző lépésben létrehozott tábla változáskövetési verziójának frissítéséhez.

   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

Telepítse a legújabb Azure PowerShell modulokat a Hogyan telepítse és konfigurálja az Azure PowerShellt útmutató utasításaiban.

Adat-előállító létrehozása

1. Adjon meg egy olyan változót, amelyet később a PowerShell-parancsokban az erőforráscsoport neveként fog használni. Másolja a következő parancsszöveget a PowerShellbe, adja meg a Azure erőforráscsoport nevét dupla idézőjelek között, majd futtassa a parancsot. Például: "adfrg"

   $resourceGroupName = "ADFTutorialResourceGroup";
   

   Ha az erőforráscsoport már létezik, előfordulhat, hogy nem kívánja felülírni. Rendeljen egy másik értéket a $resourceGroupName változóhoz, majd futtassa újra a parancsot

2. Adjon meg egy változót az adat-előállító helyéhez:

   $location = "East US"
   

3. A Azure erőforráscsoport létrehozásához futtassa a következő parancsot:

   New-AzResourceGroup $resourceGroupName $location
   

   Ha az erőforráscsoport már létezik, előfordulhat, hogy nem kívánja felülírni. Rendeljen egy másik értéket a $resourceGroupName változóhoz, majd futtassa újra a parancsot.

4. Adjon meg egy változót az adat-előállító nevéhez.

   Fontos

   Frissítse az adat-előállító nevét, hogy globálisan egyedi legyen.

   $dataFactoryName = "IncCopyChgTrackingDF";
   

5. Az adat-előállító létrehozásához futtassa a következő Set-AzDataFactoryV2 parancsmagot:

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

Vegye figyelembe az alábbiakat:

• A Azure adat-előállító nevének globálisan egyedinek kell lennie. Ha a következő hibaüzenetet kapja, módosítsa a nevet, majd próbálkozzon újra.

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

• Data Factory-példányok létrehozásához ahhoz, hogy be tudjon jelentkezni az Azure-ba, a felhasználói fióknak a contributor vagy owner szerepkörök valamelyikének tagja kell legyen, vagy az Azure előfizetésének az adminisztrátora.

• Azoknak a Azure régióknak a listájához, amelyekben a Data Factory jelenleg elérhető, jelölje ki az Önt érdeklő régiókat az alábbi lapon, majd bontsa ki a Analytics elemet a Data Factory: Régiók szerint elérhető termékek. A data factory által használt adattárak (Azure Storage, Azure SQL Database stb.) és a data factory által használt számítások (HDInsight stb.) más régiókban is lehetnek.

Társított szolgáltatások létrehozása

Társított szolgáltatásokat hoz létre egy adat-előállítóban az adattárak és a számítási szolgáltatások adat-előállítóval történő társításához. Ebben a szakaszban kapcsolt szolgáltatásokat hoz létre az Azure Storage-fiókjához és az Azure SQL Database-ben található adatbázisához.

Azure Storage-hoz kapcsolt szolgáltatás létrehozása.

Ebben a lépésben összekapcsolja a Azure Storage-fiókját az adat-előállítóval.

1. Hozzon létre egy AzureStorageLinkedService.json nevű JSON-fájlt a C:\ADFTutorials\IncCopyChangeTrackingTutorial mappában a következő tartalommal (ha még nem létezik a mappa, hozza létre). A fájl mentése előtt cserélje le <accountName>, <accountKey> a Azure tárfiók nevére és kulcsára.

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

2. Az Azure PowerShell-ban váltson a C:\ADFTutorials\IncCopyChangeTrackingTutorial mappára.

3. Futtassa a Set-AzDataFactoryV2LinkedService parancsmagot a társított szolgáltatás létrehozásához: AzureStorageLinkedService. A következő példában a ResourceGroupName és a DataFactoryName paraméter értékeit fogja megadni.

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

   Itt látható a minta kimenete:

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

Azure SQL Database társított szolgáltatás létrehozása.

Ebben a lépésben az adatbázist az adat-előállítóhoz csatolja.

1. Hozzon létre egy AzureSQLDatabaseLinkedService.json nevű JSON fájlt a következő mappában: C:\ADFTutorials\IncCopyChangeTrackingTutorial. Mielőtt elmenti a fájlt, cserélje ki a <your-server-name> és <your-database-name> helyőrzőket a saját szerverének és adatbázisának nevére. Az Azure SQL Servert úgy kell konfigurálnia, hogy hozzáférést adjon az adatgyár felügyelt identitásához.

   {
   "name": "AzureSqlDatabaseLinkedService",
   "properties": {
           "type": "AzureSqlDatabase",
           "typeProperties": {
               "connectionString": "Server=tcp:<your-server-name>.database.windows.net,1433;Database=<your-database-name>;"
           },
           "authenticationType": "ManagedIdentity",
           "annotations": []
       }
   }
   

2. A Azure PowerShell futtassa a Set-AzDataFactoryV2LinkedService parancsmagot a társított szolgáltatás létrehozásához: AzureSQLDatabaseLinkedService.

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

   Itt látható a minta kimenete:

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

Adatkészletek létrehozása

Ebben a lépésben adatkészleteket hoz létre, amelyek az adatforrást és az adatcélpontot képviselik. és a SYS_CHANGE_VERSION tárolási helyét.

Forrásadatkészlet létrehozása

Ebben a lépésben egy adatkészletet hoz létre, amely a forrásadatokat jelöli.

1. Hozzon létre egy SourceDataset.json nevű JSON-fájlt ugyanebben a mappában az alábbi tartalommal:

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

2. Futtassa a Set-AzDataFactoryV2Dataset parancsmagot az adatkészlet létrehozásához: SourceDataset

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

   Itt látható a parancsmag mintakimenete:

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

Adatkimenet adatkészlet létrehozása

Ebben a lépésben egy adatkészletet hoz létre, amely a forrásadattárból másolt adatokat jelöli.

1. Hozzon létre egy SinkDataset.json nevű JSON-fájlt ugyanebben a mappában az alábbi tartalommal:

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

   Az előfeltételek részeként hozza létre az adftutorial tárolót az Azure Blob Storage-ban. Ha még nem létezik, hozza létre a tárolót, vagy állítsa be egy meglévő tároló nevét. Ebben az oktatóanyagban a kimeneti fájl neve dinamikusan jön létre a következő kifejezéssel: @CONCAT("Incremental-", pipeline(). RunId, '.txt').

2. Futtassa a Set-AzDataFactoryV2Dataset parancsmagot az adatkészlet létrehozásához: SinkDataset

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

   Itt látható a parancsmag mintakimenete:

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

Változáskövetési adatkészlet létrehozása

Ebben a lépésben egy adatkészletet hozunk létre a változáskövetés verziószámának tárolásához.

1. Hozzon létre egy ChangeTrackingDataset.json nevű JSON-fájlt ugyanebben a mappában az alábbi tartalommal:

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

   Ön az előfeltételek részeként egy table_store_ChangeTracking_version nevű táblát hoz létre.

2. Futtassa a Set-AzDataFactoryV2Dataset parancsmagot az adathalmaz létrehozásához: ChangeTrackingDataset

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

   Itt látható a parancsmag mintakimenete:

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

Folyamat létrehozása teljes másolat készítéséhez

Ebben a lépésben létrehoz egy másolási tevékenységet tartalmazó folyamatot, amely a teljes adatot a forrásadattárból (Azure SQL Database) a céladattárba (Azure Blob Storage) másolja.

1. Hozzon létre egy FullCopyPipeline.json nevű JSON-fájlt ugyanebben a mappában az alábbi tartalommal:

   {
       "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. Futtassa a Set-AzDataFactoryV2Pipeline parancsmagot a folyamat létrehozásához: FullCopyPipeline.

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

   Itt látható a minta kimenete:

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

A teljes másolási folyamat futtatása

Futtassa a folyamatot: FullCopyPipeline az Invoke-AzDataFactoryV2Pipeline parancsmaggal.

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

A teljes másolási folyamat megfigyelése

1. Jelentkezzen be Azure portálra.

2. Kattintson a Minden szolgáltatás elemre, végezzen keresést a data factories kulcsszóval, és válassza az Adat-előállítók lehetőséget.

   

3. Keresse meg az adat-előállítóját az adat-előállítók listájában, és kattintson rá az Adat-előállító lap megnyitásához.

   

4. Az Adat-előállító lapon kattintson a Figyelés és felügyelet csempére.

   

5. A adatintegráció alkalmazás egy külön lapon indul el. Megtekintheti az összes folyamatfuttatást és azok állapotát. Kérjük, vegye észre, hogy a következő példában a pipeline futtatásának állapota Sikeres. A Paraméterek oszlopban található hivatkozásra kattintva megtekintheti a folyamatnak átadott paramétereket. Hiba esetén egy hivatkozás jelenik meg a Hiba oszlopban. Kattintson a Műveletek oszlopban található hivatkozásra.

   

6. Ha a Műveletek oszlopban található hivatkozásra kattint, megjelenik a következő lap, amely megmutatja a folyamat összes tevékenységfuttatását.

   

7. A Folyamatfuttatások nézetre való visszaváltáshoz kattintson a Folyamatok elemre a képen látható módon.

Az eredmények áttekintése

Egy incremental-<GUID>.txt nevű fájl található az incchgtracking nevű tároló adftutorial mappájában.

A fájlnak az adatbázisból származó adatokkal kell rendelkeznie:

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

További adatok hozzáadása a forrástáblához

Futtassa a következő lekérdezést az adatbázison egy rekord hozzáadásához és egy rekord frissítéséhez.

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


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

Pipeline létrehozása a delta másolathoz

Ebben a lépésben a következő tevékenységeket tartalmazó folyamatot fog létrehozni, majd időszakosan futtatni. A lekérdezési műveletek lekérik a régi és az új SYS_CHANGE_VERSION-t az Azure SQL Database-ből, és átadják a másolási tevékenységnek. A copy tevékenység a beszúrt/frissített/törölt adatokat a két SYS_CHANGE_VERSION érték között másolja az Azure SQL Database-ből az Azure Blob Storage-be. A tárolt eljárási tevékenység frissíti a SYS_CHANGE_VERSION értékét a következő folyamatfuttatáshoz.

1. Hozzon létre egy IncrementalCopyPipeline.json nevű JSON-fájlt ugyanebben a mappában az alábbi tartalommal:

   {
       "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. Futtassa a Set-AzDataFactoryV2Pipeline parancsmagot a folyamat létrehozásához: FullCopyPipeline.

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

   Itt látható a minta kimenete:

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

A növekményes másolási folyamat futtatása

Futtassa a folyamatotervet: IncrementalCopyPipeline az Invoke-AzDataFactoryV2Pipeline parancsmaggal.

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

A növekményes másolási folyamat figyelése

1. Az adatintegrációs alkalmazásban frissítse a Folyamatfuttatások nézetet. Ellenőrizze, hogy az IncrementalCopyPipeline szerepel-e a listában. Kattintson a Műveletek oszlopban található hivatkozásra.

   

2. Ha a Műveletek oszlopban található hivatkozásra kattint, megjelenik a következő lap, amely megmutatja a folyamat összes tevékenységfuttatását.

   

3. A Folyamatfuttatások nézetre való visszaváltáshoz kattintson a Folyamatok elemre a képen látható módon.

Az eredmények áttekintése

A második fájl az incchgtracking nevű tároló adftutorial mappájában található.

A fájlnak csak az adatbázis különbözeti adataival kell rendelkeznie. Az U karaktert tartalmazó rekord a frissített sor az adatbázisban, az I karaktert tartalmazó rekord pedig a hozzáadott sor.

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

Az első három oszlop a data_source_table táblából származó módosított adatokat tartalmazza. Az utolsó két oszlop a változáskövetési rendszer táblájából származó metaadatokat tartalmazza. A negyedik oszlop az egyes módosított sorokra vonatkozó SYS_CHANGE_VERSION értéket tartalmazza. Az ötödik oszlop a műveletet tartalmazza: U = frissítés, I = beszúrás. A változáskövetési adatokkal kapcsolatos információért lásd: CHANGETABLE.

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

Kapcsolódó tartalom

A következő oktatóanyagban megismerheti, hogyan másolhat új és módosított fájlokat csak a LastModifiedDate alapján:

Új fájlok másolása lastmodifieddate szerint