Kopiera data inkrementellt från Azure SQL-databasen till Blob Storage genom att använda förändringsspårning i Azure-portalen.

GÄLLER FÖR: Azure Data Factory Azure Synapse Analytics

I en dataintegrationslösning är inkrementell laddning av data efter initial dataladdning ett mycket använt scenario. De ändrade uppgifterna inom en period i din källdataförvaring kan enkelt delas upp (till exempel, LastModifyTime, CreationTime). Men i vissa fall finns det inget tydligt sätt att identifiera delta-datan från den senaste gången du bearbetade datan. Du kan använda förändringsspårningstekniken som stöds av datalager såsom Azure SQL Database och SQL Server för att identifiera deltadata.

Den här handledningen beskriver hur man använder Azure Data Factory med ändringsspårning för att inkrementellt ladda delta data från Azure SQL Database till Azure Blob Storage. Mer information om spårning av ändringar finns i Ändringsspårning i SQL Server.

Du utför följande steg i denna handledning:

Övergripande lösning

I den här handledningen skapar du två pipelines som utför följande operationer.

1. Initial loading of historical data: Du skapar en pipeline med en kopieringsaktivitet som kopierar hela datan från källdatalagret (Azure SQL Database) till destinationsdatalagret (Azure Blob Storage).

   1. Aktivera ändringsspårningsteknik i käll-databasen i Azure SQL Database.
   2. Hämta det initiala värdet av SYS_CHANGE_VERSION i databasen som en baslinje för att fånga förändrade data.
   3. Ladda den fullständiga datan från källdatabasen till Azure Blob Storage.

   

2. Inkrementell laddning av deltadata enligt schema: Du skapar ett pipeline med följande aktiviteter och kör det regelbundet.

   1. Skapa två uppslag aktiviteter för att få de gamla och nya SYS_CHANGE_VERSION värdena från Azure SQL Database.

   2. Skapa en kopieringsaktivitet för att kopiera de insatta, uppdaterade eller raderade data (delta-data) mellan de två SYS_CHANGE_VERSION värdena från Azure SQL Database till Azure Blob Storage.

      Du laddar deltamaterialet genom att sammanfoga primärnycklarna för ändrade rader (mellan två SYS_CHANGE_VERSION värden) från sys.change_tracking_tables med data i källtabellen, och sedan flyttar du deltamaterialet till destinationen.

   3. Skapa en lagrad proceduraktivitet för att uppdatera värdet på SYS_CHANGE_VERSION inför nästa körning av pipelinen.

   

Förutsättningar

• Azure-prenumeration. Om du inte har ett, skapa ett gratis konto innan du börjar.
• Azure SQL Database. Du använder en databas i Azure SQL Database som källa för datalagring. Om du inte har en, se Skapa en databas i Azure SQL Database för steg hur du skapar den.
• Azure Storage-konto. Du använder Blob Storage som mål datalagring. Om du inte har ett Azure-lagringskonto, se Skapa ett lagringskonto för instruktioner om hur du skapar ett. Skapa en behållare med namnet adftutorial.

Skapa en datakälltabell i Azure SQL Database

1. Öppna SQL Server Management Studio och anslut till SQL Database.

2. I Serverutforskaren, högerklicka på din databas och välj sedan Ny fråga.

3. Kör följande SQL-kommando mot din databas för att skapa en tabell som heter data_source_table som datalagringskälla.

   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. Aktivera förändringsspårning i din databas och källtabell (data_source_table) genom att köra följande SQL-fråga.

   Anteckning / Notis

   • Ersätt <your database name> med namnet på databasen i Azure SQL-databasen som har data_source_table.
   • Den ändrade datan sparas i två dagar i det aktuella exemplet. Om du laddar de ändrade uppgifterna var tredje dag eller mer, ingår inte vissa ändrade uppgifter. Du måste antingen ändra värdet på CHANGE_RETENTION till ett större tal eller säkerställa att tidsperioden för att ladda de ändrade uppgifterna är inom två dagar. För mer information, se Enable change tracking for a 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. Skapa en ny tabell och butik som kallas ChangeTracking_version med ett standardvärde genom att köra följande fråga:

   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)
   

   Anteckning / Notis

   Om data inte ändras efter att du aktiverar ändringsspårning för SQL-databasen, är värdet av ändringsspårningsversionen 0.

6. Kör följande fråga för att skapa en lagrad procedur i din databas. Pipelines anropar denna lagrade procedur för att uppdatera versionsnumret för förändringsspårning i tabellen du skapade i föregående steg.

   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    
   

Skapa en datafabrik

1. Öppna webbläsaren Microsoft Edge eller Google Chrome. För närvarande stöds användargränssnittet (UI) för Data Factory endast av dessa webbläsare.

2. På Azure-portalen, i den vänstra menyn, välj Skapa en resurs.

3. Välj Analytics>Data Factory :

   

4. På sidan Ny datafabrik anger du ADFTutorialDataFactory som namn.

   Namnet på dataverket måste vara globalt unikt. Om du får ett felmeddelande som säger att namnet du valde inte är tillgängligt, ändra namnet (till exempel till yournameADFTutorialDataFactory) och försök skapa datafabriken igen. För mer information, se Azure Data Factory-namngivningsregler.

5. Välj den Azure-abonnemang där du vill skapa datafabriken.

6. För Resursgrupp, ta ett av följande steg:

   • Välj Använd befintlig och välj sedan en befintlig resursgrupp från rullgardinsmenyn.
   • Välj Skapa ny, och ange sedan namnet på en resursgrupp.

   För att lära dig mer om resursgrupper, se Använda resursgrupper för att hantera dina Azure-resurser.

7. För Version, välj V2.

8. För Region, välj regionen för datafabriken.

   Rullgardinslistan visar endast platser som stöds. Databutiker (t.ex. Azure Storage och Azure SQL Database) och beräknar (t.ex. Azure HDInsight) som en datafabrik använder kan finnas i andra regioner.

9. Välj Nästa: Git-konfiguration. Ställ in arkivet genom att följa instruktionerna i Konfigurationsmetod 4: Under fabrikens skapande, eller markera kryssrutan Konfigurera Git senare.

10. Välj Förhandsgranska + skapa.

11. Välj Skapa.

    På instrumentpanelen visar Deploying Data Factory-rutan statusen.

    

12. När skapandet är klart visas Datafabrik-sidan. Välj plattan Launch studio för att öppna Azure Data Factory UI i en separat flik.

Skapa länkade tjänster

Du skapar kopplade tjänster i en datafabrik för att länka dina datalager och beräkningstjänster till datafabriken. I det här avsnittet skapar du länktjänster till ditt Azure-lagringskonto och din databas i Azure SQL Database.

Skapa en länkad Azure Storage-tjänst

För att länka ditt lagringskonto till datafabriken:

1. I Data Factory UI, på fliken Hantera, under Anslutningar, välj Länkade tjänster. Välj sedan + Ny eller knappen Skapa länkad tjänst.
2. I fönstret New Linked Service, välj Azure Blob Storage, och välj sedan Continue.
3. Ange följande information:
   1. För Namn anger du AzureStorageLinkedService.
   2. För Anslut via integration runtime, välj integration runtime.
   3. För Autentiseringstyp, välj en autentiseringsmetod.
   4. För namn på lagringskonto, välj ditt Azure-lagringskonto.
4. Välj Skapa.

Skapa en länkad tjänst för Azure SQL-databas

För att koppla din databas till datafabriken:

1. I Data Factory UI, på fliken Hantera, under Anslutningar, välj Länkade tjänster. Välj sedan + Ny.

2. I fönstret Ny länkad tjänst väljer du Azure SQL Database och sedan Fortsätt.

3. Ange följande information:

   1. För Name, ange AzureSqlDatabaseLinkedService.
   2. För Servernamn, välj din server.
   3. För Databasnamn, välj din databas.
   4. För Autentiseringstyp, välj en autentiseringsmetod. Denna handledning använder SQL-autentisering för demonstration.
   5. För User name, skriv in användarens namn.
   6. För lösenord, ange ett lösenord för användaren. Eller, ange informationen för Azure Key Vault - AKV länkad tjänst, hemligt namn och hemlig version.

4. Välj Testa anslutning för att testa anslutningen.

5. Välj Skapa för att skapa den länkade tjänsten.

   

Skapa datamängder

I det här avsnittet skapar du datamängder för att representera datakällan och datadestinationen, tillsammans med platsen för att lagra SYS_CHANGE_VERSION-värdena.

Skapa en datamängd för att representera källdata

1. I Data Factory UI, på fliken Author, välj plustecknet (+). Välj sedan Dataset, eller välj ellipsen för datasetåtgärder.

   

2. Välj Azure SQL Database, och välj sedan Fortsätt.

3. I Set Properties-fönstret, följ dessa steg:

   1. För Namn anger du SourceDataset.
   2. För Länkad tjänst väljer du AzureSqlDatabaseLinkedService.
   3. För Tabellnamn, välj dbo.data_source_table.
   4. För Importera schema, välj alternativet Från anslutning/butik.
   5. Välj OK.

   

Skapa en datamängd för att representera data kopierad till den mottagande datalagringen.

I den följande proceduren skapar du en datamängd för att representera den data som kopieras från källdataförvaret. Du skapade adftutorial-behållaren i Azure Blob Storage som en del av förutsättningarna. Skapa behållaren om den inte finns, eller ställ in den på namnet på en befintlig. I den här handledningen genereras namnet på utdatafilen dynamiskt från uttrycket @CONCAT('Incremental-', pipeline().RunId, '.txt').

1. I användargränssnittet för Data Factory, på fliken Författare, välj +. Välj sedan Dataset, eller välj ellipsen för datasetåtgärder.

   

2. Välj Azure Blob Storage, och välj sedan Fortsätt.

3. Välj dataformatet som DelimitedText, och välj sedan Fortsätt.

4. I fönstret Ange egenskaper, vidta följande steg:

   1. För Namn, ange SinkDataset.
   2. För ansluten tjänst, välj AzureBlobStorageLinkedService.
   3. För Filsökväg anger du adftutorial/incchgtracking.
   4. Välj OK.

5. När datasetet visas i trädperspektivet, gå till fliken Anslutning och välj textrutan Filnamn. När alternativet Lägg till dynamiskt innehåll visas, välj det.

   

6. Pipelineuttrycksbyggare-fönstret visas. Klistra in @concat('Incremental-',pipeline().RunId,'.csv') i textrutan.

7. Välj OK.

Skapa ett dataset för att representera data för ändringsspårning

I följande procedur skapar du en datamängd för lagring av versionsnummer för ändringsspårning. Du skapade table_store_ChangeTracking_version-tabellen som en del av förutsättningarna.

1. I Data Factory UI, på fliken Författare, välj +, och välj sedan Dataset.
2. Välj Azure SQL Database, och välj sedan Fortsätt.
3. I fönstret Ställ in egenskaper, följ dessa steg:
   1. För Namn, ange ChangeTrackingDataset.
   2. För Länkad tjänst väljer du AzureSqlDatabaseLinkedService.
   3. För Tabellnamn, välj dbo.table_store_ChangeTracking_version.
   4. För Importera schema, välj alternativet Från anslutning/butik.
   5. Välj OK.

Skapa en pipeline för hela kopian

I följande procedur skapar du ett pipeline med en kopieringsaktivitet som kopierar hela datan från källdatalagret (Azure SQL Database) till måldatalagret (Azure Blob Storage):

1. I Data Factory-gränssnittet, på fliken Author, välj +, och välj sedan Pipeline>Pipeline.

   

2. En ny flik visas för att konfigurera pipelinen. "Pipeline-flödet visas också i trädvyn." I Egenskaper-fönstret, ändra namnet på pipelinen till FullCopyPipeline.

3. I verktygslådan Aktiviteter expanderar du Flytta & omforma. Vidta en av följande åtgärder:

   • Dra kopiaaktiviteten till pipelinedesignerns yta.
   • I sökfältet under Aktiviteter, sök efter kopieringsaktiviteter för data, och ställ sedan in namnet till FullCopyActivity.

4. Växla till fliken Source. För Source Dataset, välj SourceDataset.

5. Växla till fliken Sink. För Sink Dataset, välj SinkDataset.

6. För att validera rördefinitionen, välj Validera på verktygsfältet. Bekräfta att det inte finns något valideringsfel. Stäng utmatningen från pipeline-validering.

7. För att publicera enheterna (länkade tjänster, dataset och pipelines), välj Publicera alla. Vänta tills du ser meddelandet Framgångsrikt publicerat.

   

8. För att se aviseringar, välj knappen Visa aviseringar.

Kör den fullständiga kopieringspipelin

1. I Data Factory UI, på verktygsfältet för pipeline, välj Lägg till utlösare och välj sedan Utlös nu.

   

2. I fönstret Pipelinekörning väljer du OK.

   

Övervaka hela kopieringskedjan

1. I Data Factory UI:n, välj fliken Övervaka. Körningen av pipelinen och dess status visas i listan. För att uppdatera listan, välj Uppdatera. Hovra över pipelinekörningen för att få alternativet Rerun eller Förbrukning.

   

2. För att se aktivitetskörningar som är associerade med pipeline-körningen, välj pipelinens namn från kolumnen Pipelinens namn. Det finns bara en aktivitet i pipeline, så det finns bara en post i listan. För att återgå till vyn för pipeline-körningar väljer du länken Alla pipeline-körningar högst upp.

Granska resultaten

Mappen incchgtracking i behållaren adftutorial innehåller en fil med namnet incremental-<GUID>.csv.

Filen bör innehålla data från din databas.

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

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

Lägg till mer data i källtabellen

Kör följande fråga mot din databas för att lägga till en rad och uppdatera en rad.

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


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

Skapa en pipeline för delta kopiering

I den följande proceduren skapar du en pipeline med aktiviteter och kör den regelbundet. När du kör pipelinen:

• Uppslagsaktiviteterna hämtar de gamla och nya SYS_CHANGE_VERSION-värdena från Azure SQL Database och skickar dem till kopieringsaktiviteten.
• Kopieringsaktiviteten kopierar de infogade, uppdaterade eller raderade data mellan de två värdena från Azure SQL Database till Azure Blob Storage.
• Aktiviteten för lagrad procedur uppdaterar värdet för nästa pipeline-körning.

1. I Data Factory-gränssnittet, växla till fliken Författare. Välj +, och välj sedan Pipeline>Pipeline.

   

2. En ny flik visas för att konfigurera pipelinen. "Pipeline-flödet visas också i trädvyn." I fönstret Egenskaper, ändra namnet på pipelinen till IncrementalCopyPipeline.

3. Expandera Generella i Aktiviteter verktygslådan. Dra sökaktiviteten till pipelinedesignerns yta, eller sök i rutan Sök aktiviteter. Ange namnet på aktiviteten till LookupLastChangeTrackingVersionActivity. Den här aktiviteten hämtar versionsinformationen för ändringsspårning som användes vid den senaste kopieringsoperationen som lagras i table_store_ChangeTracking_version-tabellen.

4. Växla till fliken Inställningar i fönstret Egenskaper. För Source Dataset väljer du ChangeTrackingDataset.

5. Dra lookup-aktiviteten från Aktiviteter-verktygslådan till pipelinedesignerns yta. Ange aktivitetens namn till LookupCurrentChangeTrackingVersionActivity. Den här aktiviteten hämtar den aktuella versionshistoriken.

6. Växla till fliken Inställningar i Egenskaper-fönstret, och följ sedan dessa steg:

   1. För Källdataset, välj SourceDataset.

   2. För Använd fråga, välj Fråga.

   3. För Fråga, ange följande SQL-fråga:

      SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion
      

   

7. I verktygslådan Aktiviteter expanderar du Flytta & omforma. Dra kopiera data-aktiviteten till pipeline-designytan. Ställ in namnet på aktiviteten till IncrementalCopyActivity. Denna aktivitet kopierar data mellan den senaste för versionskontroll och den nuvarande versionskontrollen till måldataförvaret.

8. Byt till fliken Källa i Egenskaper-fönstret och följ sedan dessa steg:

   1. För Källdataset, välj SourceDataset.

   2. För Använd fråga, välj Fråga.

   3. För Fråga, ange följande SQL-fråga:

      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}
      

   

9. Växla till fliken Sink. För Sink Dataset, välj SinkDataset.

10. Anslut båda uppslagsaktiviteterna till kopieringsaktiviteten, en i taget. Dra det gröna knappen som är kopplat till uppslagaktiviteten till kopieringsaktiviteten.

11. Dra aktiviteten för lagrade procedurer från verktygslådan Aktiviteter till ytan för piplindesignern. Ange aktivitetens namn till StoredProceduretoUpdateChangeTrackingActivity. Denna aktivitet uppdaterar versionen för ändringsspårning i table_store_ChangeTracking_version-tabellen.

12. Byt till fliken Inställningar och följ sedan dessa steg:

    1. För Länkad tjänst väljer du AzureSqlDatabaseLinkedService.
    2. För lagrat procedurnamn, välj Update_ChangeTracking_Version.
    3. Välj Importera.
    4. I avsnittet Lagrade procedurparametrar, ange följande värden för parametrarna:

    Namn	Typ	Värde
    CurrentTrackingVersion	Int64	@{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
    TableName	Sträng	@{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}

    

13. Anslut kopieringsaktiviteten till procedurbegäraraktiviteten. Dra det gröna knappen som är fäst vid kopieringsaktiviteten till lagrade proceduraktiviteten.

14. Välj Validate på verktygsfältet. Bekräfta att det inte finns några valideringsfel. Stäng fönstret Pipeline-valideringsrapport .

15. Publicera enheter (länkade tjänster, dataset och pipelines) till Data Factory-tjänsten genom att välja knappen Publicera alla. Vänta tills meddelandet Publishing succeeded visas.

    

Kör den inkrementella kopieringspipeline

1. Välj Lägg till trigger på verktygsfältet för pipelinen och välj sedan Trigger nu.

   

2. I fönstret Pipelinekörning väljer du OK.

Övervaka den inkrementella kopieringspipeline

1. Välj fliken Monitor. Körningen av pipeline och dess status visas i listan. För att uppdatera listan, välj Uppdatera.

   

2. För att se aktivitetkörningar associerade med pipeline-körningen, välj länken IncrementalCopyPipeline i kolumnen Pipeline-namn. Aktiviteternas körningar visas i en lista.

   

Granska resultaten

Den andra filen visas i incchgtracking-mappen i adftutorial-containern.

Filen ska endast innehålla skillnadsdata från din databas. Posten med U är den uppdaterade raden i databasen, och I är den enda tillagda raden.

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

De första tre kolumnerna är ändrade data från data_source_table. De två sista kolumnerna är metadata från tabellen för ändringsspårningssystemet. Den fjärde kolumnen är SYS_CHANGE_VERSION-värdet för varje ändrad rad. Den femte kolumnen är operationen: U = uppdatera, I = infoga. För mer information om spårningsinformationen för ändringar, se CHANGETABLE.

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

Relaterat innehåll

Fortsätt till nästa handledning för att lära dig hur du kopierar endast nya och ändrade filer, baserat på LastModifiedDate.