Mønster for trinnvis innsamling av data med Dataflyt gen2

Denne opplæringen tar 15 minutter, og beskriver hvordan du trinnvis samler inn data i et lakehouse ved hjelp av Dataflow Gen2.

Trinnvis innsamling av data i et datamål krever en teknikk for å laste inn bare nye eller oppdaterte data i datamålet. Denne teknikken kan gjøres ved hjelp av en spørring for å filtrere dataene basert på datamålet. Denne opplæringen viser hvordan du oppretter en dataflyt for å laste inn data fra en OData-kilde til et lakehouse, og hvordan du legger til en spørring i dataflyten for å filtrere dataene basert på datamålet.

Trinnene på høyt nivå i denne opplæringen er som følger:

• Opprett en dataflyt for å laste inn data fra en OData-kilde til et lakehouse.
• Legg til en spørring i dataflyten for å filtrere dataene basert på datamålet.
• (Valgfritt) laster inn data på nytt ved hjelp av notatblokker og datasamlebånd.

Forutsetning

Du må ha et Microsoft Fabric-aktivert arbeidsområde. Hvis du ikke allerede har et, kan du se Opprette et arbeidsområde. Opplæringen forutsetter også at du bruker diagramvisningen i Dataflyt gen2. Hvis du vil kontrollere om du bruker diagramvisningen, går du til Vis på det øverste båndet og kontrollerer at diagramvisning er valgt.

Opprette en dataflyt for å laste inn data fra en OData-kilde til et lakehouse

I denne delen oppretter du en dataflyt for å laste inn data fra en OData-kilde til et lakehouse.

1. Opprett et nytt lakehouse i arbeidsområdet.

   

2. Opprett en ny Dataflyt gen2 i arbeidsområdet.

   

3. Legg til en ny kilde i dataflyten. Velg OData-kilden, og skriv inn følgende URL-adresse: https://services.OData.org/V4/Northwind/Northwind.svc

   

   

   

4. Velg Ordrer-tabellen, og velg Neste.

   

5. Velg følgende kolonner du vil beholde:

   • OrderID
   • CustomerID
   • EmployeeID
   • OrderDate
   • RequiredDate
   • ShippedDate
   • ShipVia
   • Freight
   • ShipName
   • ShipAddress
   • ShipCity
   • ShipRegion
   • ShipPostalCode
   • ShipCountry

   

   

6. Endre datatype OrderDatefor , RequiredDateog ShippedDate til datetime.

   

7. Konfigurer datamålet til lakehouse ved hjelp av følgende innstillinger:

   • Datamål: Lakehouse
   • Lakehouse: Velg lakehouse du opprettet i trinn 1.
   • Nytt tabellnavn: Orders
   • Oppdateringsmetode: Replace

   

   

   

8. velg Neste , og publiser dataflyten.

   

Du har nå opprettet en dataflyt for å laste inn data fra en OData-kilde til et lakehouse. Denne dataflyten brukes i neste del til å legge til en spørring i dataflyten for å filtrere dataene basert på datamålet. Deretter kan du bruke dataflyten til å laste inn data på nytt ved hjelp av notatblokker og datasamlebånd.

Legge til en spørring i dataflyten for å filtrere dataene basert på datamålet

Denne delen legger til en spørring i dataflyten for å filtrere dataene basert på dataene i målsjøen. Spørringen får maksimalt i OrderID lakehouse i begynnelsen av dataflytoppdateringen og bruker den maksimale OrderId til å bare få ordrene med en høyere OrderId fra til kilde for å tilføye til datamålet. Dette forutsetter at ordrer legges til kilden i stigende rekkefølge av OrderID. Hvis dette ikke er tilfelle, kan du bruke en annen kolonne til å filtrere dataene. Du kan for eksempel bruke OrderDate kolonnen til å filtrere dataene.

Merk

OData-filtre brukes i Fabric etter at dataene er mottatt fra datakilden, men for databasekilder som SQL Server brukes filteret i spørringen som sendes til serverdeldatakilden, og bare filtrerte rader returneres til tjenesten.

1. Når dataflyten er oppdatert, åpner du dataflyten du opprettet i forrige del.

   

2. Opprett en ny spørring med navnet IncrementalOrderID og hent data fra Ordrer-tabellen i lakehouse du opprettet i forrige del.

   

   

   

   

   

3. Deaktiver oppsamling av denne spørringen.

   

4. Høyreklikk kolonnen i forhåndsvisningen av OrderID data, og velg Drill ned.

   

5. Velg Listeverktøy -Statistikk> fra båndet.

   

Du har nå en spørring som returnerer maksimal ordre-ID i lakehouse. Denne spørringen brukes til å filtrere dataene fra OData-kilden. Den neste delen legger til en spørring i dataflyten for å filtrere dataene fra OData-kilden basert på maksimal OrdreID i lakehouse.

1. Gå tilbake til Ordrer-spørringen, og legg til et nytt trinn for å filtrere dataene. Bruk følgende innstillinger:

   • Kolonne: OrderID
   • Operasjon: Greater than
   • Verdi: parameter IncrementalOrderID

   

   

2. Tillat kombinasjon av data fra OData-kilden og lakehouse ved å bekrefte følgende dialogboks:

   

3. Oppdater datamålet for å bruke følgende innstillinger:

   • Oppdateringsmetode: Append

   

   

   

4. Publiser dataflyten.

   

Dataflyten inneholder nå en spørring som filtrerer dataene fra OData-kilden basert på maksimal OrdreID i lakehouse. Dette betyr at bare nye eller oppdaterte data lastes inn i lakehouse. Den neste delen bruker dataflyten til å laste inn data på nytt ved hjelp av notatblokker og datasamlebånd.

(Valgfritt) laster inn data på nytt ved hjelp av notatblokker og datasamlebånd

Du kan eventuelt laste inn bestemte data på nytt ved hjelp av notatblokker og datasamlebånd. Med egendefinert python-kode i notatblokken fjerner du de gamle dataene fra lakehouse. Når du oppretter et datasamlebånd der du først kjører notatblokken og kjører dataflyten sekvensielt, laster du dataene fra OData-kilden inn i lakehouse. Notatblokker støtter flere språk, men denne opplæringen bruker PySpark. Pyspark er en Python-API for Spark og brukes i denne opplæringen til å kjøre Spark SQL-spørringer.

1. Opprett en ny notatblokk i arbeidsområdet.

   

2. Legg til følgende PySpark-kode i notatblokken:

   ### Variables
   LakehouseName = "YOURLAKEHOUSE"
   TableName = "Orders"
   ColName = "OrderID"
   NumberOfOrdersToRemove = "10"
   
   
   ### Remove Old Orders
   Reload = spark.sql("SELECT Max({0})-{1} as ReLoadValue FROM {2}.{3}".format(ColName,NumberOfOrdersToRemove,LakehouseName,TableName)).collect()
   Reload = Reload[0].ReLoadValue
   spark.sql("Delete from {0}.{1} where {2} > {3}".format(LakehouseName, TableName, ColName, Reload))
   

3. Kjør notatblokken for å bekrefte at dataene er fjernet fra lakehouse.

4. Opprett et nytt datasamlebånd i arbeidsområdet.

   

5. Legg til en ny notatblokkaktivitet i datasamlebåndet, og velg notatblokken du opprettet i forrige trinn.

   

   

6. Legg til en ny dataflytaktivitet i datasamlebåndet, og velg dataflyten du opprettet i forrige del.

   

   

7. Koble notatblokkaktiviteten til dataflytaktiviteten med en vellykket utløser.

   

8. Lagre og kjør datasamlebåndet.

   

Du har nå et datasamlebånd som fjerner gamle data fra lakehouse og laster dataene fra OData-kilden inn i lakehouse. Med dette oppsettet kan du laste inn dataene fra OData-kilden regelmessig i lakehouse.