Dela via

Läs in data inkrementellt från Data Warehouse till Lakehouse

Den här handledningen visar hur du endast kopierar nya eller uppdaterade data från ditt informationslager till ett Lakehouse. Den här metoden kallas inkrementell inläsning och är användbar när du vill behålla dina data up-to-date utan att kopiera allt varje gång.

Här är lösningens design på hög nivå:

Diagram som visar inkrementell inläsning av datalogik.

  1. Välj en kolumn för vattenstämplar. Välj en kolumn i källtabellen som hjälper dig att spåra nya eller ändrade poster. Den här kolumnen innehåller vanligtvis värden som ökar när rader läggs till eller uppdateras (till exempel en tidsstämpel eller ett ID). Vi använder det högsta värdet i den här kolumnen som vår "referenspunkt" för att veta var vi slutade.

  2. Konfigurera en tabell för att lagra ditt senaste vattenstämpelvärde.

  3. Skapa en pipeline som gör följande:

    Pipelinen innehåller följande aktiviteter:

    • Två uppslagsaktiviteter. Den första får det sista vattenstämpelvärdet (där vi slutade förra gången). Den andra får det nya vattenstämpelvärdet (där vi stoppar den här gången). Båda värdena överförs till kopieringsaktiviteten.
    • En kopieringsaktivitet som hittar rader där värdet för vattenstämpelkolumnen ligger mellan de gamla och nya vattenstämplarna. Sedan kopieras dessa data från ditt informationslager till Lakehouse som en ny fil.
    • En lagrad proceduraktivitet som sparar det nya vattenstämpelvärdet så att nästa körning av pipeline vet var den ska börja.

Förutsättningar

  • Informationslager. Du använder informationslagret som källdatalager. Om du inte har någon kan du läsa mer i Skapa ett informationslager .
  • Lakehouse. Du använder Lakehouse som måldatalager. Om du inte har någon, se Skapa en Lakehouse för anvisningar.
    • Skapa en mapp med namnet IncrementalCopy för att lagra dina kopierade data.

Förbereda källan

Nu ska vi konfigurera tabellerna och den lagrade proceduren som du behöver i informationslagret innan du konfigurerar den inkrementella kopieringspipelinen.

1. Skapa en datakällatabell i informationslagret

Kör följande SQL-kommando i informationslagret för att skapa en tabell med namnet data_source_table som källtabell. Vi använder detta som exempeldata för den inkrementella kopian.

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

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

Data i källtabellen ser ut så här:

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

I den här tutorialen använder vi LastModifytime som vattenmärkeskolumn.

2. Skapa en annan tabell i informationslagret för att lagra det sista vattenstämpelvärdet

  1. Kör följande SQL-kommando i informationslagret för att skapa en tabell med namnet watermarktable för att lagra det sista vattenstämpelvärdet:

    create table watermarktable
(
TableName varchar(255),
WatermarkValue DATETIME2(6),
);

  2. Ange standardvärdet för den senaste vattenstämpeln med källtabellens namn. I den här självstudien är tabellnamnet data_source_table och vi anger standardvärdet till 1/1/2010 12:00:00 AM.

    INSERT INTO watermarktable
VALUES ('data_source_table','1/1/2010 12:00:00 AM')

  3. Kontrollera data i watermarktabell.

    Select * from watermarktable

    Utdata:

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

3. Skapa en lagrad procedur i informationslagret

Kör följande kommando för att skapa en lagrad procedur i informationslagret. Den här lagrade proceduren uppdaterar det senaste vattenstämpelvärdet efter varje pipelinekörning.

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

BEGIN

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

END

Konfigurera en pipeline för inkrementell kopiering

Steg 1: Skapa en pipeline

  1. Gå till Power BI.

  2. Välj Power BI-ikonen längst ned till vänster på skärmen och välj sedan Infrastruktur.

  3. Välj Min arbetsyta för att öppna arbetsytan Fabric.

  4. Välj + Nytt objekt, välj Pipeline och ange sedan ett pipelinenamn för att skapa en ny pipeline.

    Skärmbild som visar den nya pipelineknappen på den nyligen skapade arbetsytan.

    Skärmbild som visar namnet på hur du skapar en ny pipeline.

Steg 2: Lägg till en uppslagsaktivitet för den senaste vattenstämpeln

I det här steget skapar du en uppslagsaktivitet för att hämta det sista vattenstämpelvärdet. Vi får det standardvärde 1/1/2010 12:00:00 AM som vi angav tidigare.

  1. Välj Pipelineaktivitet och välj Uppslag från rullgardinsmenyn.

  2. Under fliken Allmänt byter du namn på aktiviteten till LookupOldWaterMarkActivity.

  3. Under fliken Inställningar konfigurerar du följande:

    • Anslutning: Under Lager väljer du Bläddra bland alla och väljer ditt informationslager i listan.
    • Använd fråga: Välj Tabell.
    • Tabell: Välj dbo.watermarktable.
    • Endast första raden: Vald.

    Skärmbild som visar den gamla uppslagsvattenstämpeln.

Steg 3: Lägg till en uppslagsaktivitet för den nya vattenstämpeln

I det här steget skapar du en uppslagsaktivitet för att hämta det nya vattenstämpelvärdet. Du använder en sökfråga för att hämta den nya vattenstämpeln från källans datatabell. Vi får det högsta värdet i kolumnen LastModifytime från data_source_table.

  1. I det övre fältet väljer du Uppslag under fliken Aktiviteter för att lägga till den andra uppslagsaktiviteten.

  2. Under fliken Allmänt byter du namn på aktiviteten till LookupNewWaterMarkActivity.

  3. Under fliken Inställningar konfigurerar du följande:

    • Anslutning: Under Informationslager väljer du Bläddra bland alla och väljer ditt informationslager i listan eller väljer ditt informationslager från Infrastrukturobjektanslutningar.

    • Använd fråga: Välj Fråga.

    • Fråga: Ange följande fråga för att välja den maximala senast ändrade tiden som den nya vattenstämpeln:

      select MAX(LastModifytime) as NewWatermarkvalue from data_source_table

    • Endast första raden: Vald.

    Skärmbild som visar ny vattenstämpel för uppslag.

Steg 4: Lägg till kopieringsaktiviteten för att kopiera inkrementella data

I det här steget lägger du till en kopieringsaktivitet för att kopiera inkrementella data mellan den senaste vattenstämpeln och den nya vattenstämpeln från ditt informationslager till Lakehouse.

  1. Välj Aktiviteter i det övre fältet och välj Kopiera data –> Lägg till på arbetsytan för att hämta kopieringsaktiviteten.

  2. Under fliken Allmänt byter du namn på den här aktiviteten till IncrementalCopyActivity.

  3. Anslut båda uppslagsaktiviteterna till kopieringsaktiviteten genom att dra den gröna knappen (Vid lyckat resultat) som är kopplad till uppslagsaktiviteterna till kopieringsaktiviteten. Släpp musknappen när du ser kantfärgen för kopieringsaktiviteten ändras till grön.

    Skärmbild som visar hur du ansluter uppslags- och kopieringsaktiviteter.

  4. Under fliken Källa konfigurerar du följande:

    • Anslutning: Under Informationslager väljer du Bläddra bland alla och väljer ditt informationslager i listan eller väljer ditt informationslager från Infrastrukturobjektanslutningar.

    • Lager: Välj ditt lager.

    • Använd fråga: Välj Fråga.

    • Fråga: Ange följande fråga för att kopiera inkrementella data mellan den senaste vattenstämpeln och den nya vattenstämpeln.

      select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'

    Skärmbild som visar kopieringskällans konfiguration.

  5. Under fliken Mål konfigurerar du följande:

    • Anslutning: Under Lakehouse väljer du Bläddra bland alla och väljer ditt sjöhus i listan eller väljer ditt lakehouse från Infrastrukturobjektanslutningar.
    • Lakehouse: Välj din Lakehouse.
    • Rotmapp: Välj Filer.
    • Filsökväg: Välj den mapp där du vill lagra dina kopierade data. Välj Bläddra för att välja din mapp. För att ange filnamnet öppnar du Lägg till dynamiskt innehåll och anger @CONCAT('Incremental-', pipeline().RunId, '.txt') i det fönster som öppnas för att skapa filnamn för dina kopierade datafiler i Lakehouse-miljön.
    • Filformat: Välj formattypen för dina data.

    Skärmbild som visar konfiguration för kopieringsdestination.

Steg 5: Lägga till en lagrad proceduraktivitet

I det här steget lägger du till en aktivitet för att uppdatera det sista markörvärdet i den lagrade proceduren inför nästa pipelinekörning.

  1. Välj Aktiviteter i det övre fältet och välj Lagrad procedur för att lägga till en lagrad proceduraktivitet.

  2. Under fliken Allmänt byter du namn på den här aktiviteten till StoredProceduretoWriteWatermarkActivity.

  3. Anslut det gröna (vid framgång) utdataflöde för kopieringsaktiviteten till aktiviteten för lagrad procedur.

  4. Under fliken Inställningar konfigurerar du följande:

    • Informationslager: Välj ditt informationslager.

    • Namn på lagrad procedur: Välj den lagrade procedur som du skapade i ditt informationslager: [dbo].[ usp_write_watermark].

    • Expandera parametrar för lagrad procedur. Om du vill ange värden för parametrarna för lagrad procedur väljer du Importera och anger följande värden för parametrarna:

      Namn Typ Värde
      SenastÄndradTid Datum/tid @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
      TabellNamn String @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

    Skärmbild som visar konfiguration av lagrad proceduraktivitet.

Steg 6: Kör pipelinen och övervaka resultatet

I det övre fältet väljer du Kör under fliken Start . Välj sedan Spara och kör. Pipelinen börjar köras och du kan övervaka den under fliken Utdata .

Skärmbild som visar resultat av pipelinekörning.

Gå till Lakehouse så ser du att datafilen finns under den mapp som du har valt. Du kan välja filen för att förhandsgranska kopierade data.

Skärmbild som visar lakehouse-data för den första körningen av rörledningen.

Skärmbild som visar förhandsversionen av Lakehouse-data för den första pipelinekörningen.

Lägg till mer data för att se resultatet av inkrementell kopiering

När du har slutfört den första pipelinekörningen ska vi lägga till mer data i källtabellen för informationslagret för att se om den här pipelinen kan kopiera dina inkrementella data.

Steg 1: Lägg till mer data i källan

Infoga nya data i informationslagret genom att köra följande fråga:

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

Uppdaterade data för data_source_table är:

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

Steg 2: Utlösa en annan pipelinekörning och övervaka resultatet

Gå tillbaka till din pipelinesida. I det övre fältet väljer du Kör under fliken Start igen. Pipelinen börjar köras och du kan övervaka den under Utdata.

Gå till Lakehouse så ser du att den nya kopierade datafilen finns under den mapp som du har valt. Du kan välja filen för att förhandsgranska kopierade data. Du kommer att se att dina inkrementella data visas i den här filen.

Skärmbild som visar lakehouse-data för den andra pipelinekörningen.

Skärmbild som visar förhandsversionen av Lakehouse-data för den andra pipelinekörningen.

Relaterat innehåll

Läs sedan mer om att kopiera från Azure Blob Storage till Lakehouse.

Kopiera från Azure Blob Storage till Lakehouse

  • Last updated on