Incrementeel gegevens uit een Azure SQL-database laden in Azure Blob Storage met de Azure-portal
VAN TOEPASSING OP:
Azure Data Factory Azure Synapse Analytics
Tip
Probeer Data Factory uit in Microsoft Fabric, een alles-in-één analyseoplossing voor ondernemingen. Microsoft Fabric omvat alles, van gegevensverplaatsing tot gegevenswetenschap, realtime analyses, business intelligence en rapportage. Meer informatie over het gratis starten van een nieuwe proefversie .
In deze zelfstudie maakt u een Azure-gegevensfactory met een pijplijn waarmee deltagegevens uit een tabel in een Azure SQL Database worden geladen naar Azure Blob Storage.
In deze zelfstudie voert u de volgende stappen uit:
- Bereid de gegevensopslag voor om de grenswaarde in op te slaan.
- Een data factory maken.
- Maak gekoppelde services.
- Maak bron-, sink- en grenswaardegegevenssets.
- Een pipeline maken.
- Voer de pijplijn uit.
- Controleer de pijplijnuitvoering.
- Resultaten controleren
- Voeg meer gegevens toe aan de bron.
- Voer de pijplijn opnieuw uit.
- Controleer de tweede pijplijnuitvoering.
- Bekijk de resultaten van de tweede uitvoering.
Overzicht
Hier volgt de diagramoplossing op hoog niveau:
Dit zijn de belangrijke stappen voor het maken van deze oplossing:
Selecteer de grenswaardekolom. Selecteer één kolom in de brongegevensopslag die kan worden gebruikt om de nieuwe of bijgewerkte records voor elke uitvoering te segmenteren. Normaal gesproken nemen de gegevens in deze geselecteerde kolom (bijvoorbeeld, last_modify_time of id) toe wanneer de rijen worden gemaakt of bijgewerkt. De maximale waarde in deze kolom wordt gebruikt als grenswaarde.
Bereid een gegevensopslag voor om de grenswaarde in op te slaan. In deze zelfstudie slaat u de grenswaarde op in een SQL-database.
Maak een pijplijn met de volgende werkstroom:
De pijplijn in deze oplossing heeft de volgende activiteiten:
- Maak twee opzoekactiviteiten. Gebruik de eerste opzoekactiviteit om de laatste grenswaarde op te halen. Gebruik de tweede opzoekactiviteit om de nieuwe grenswaarde op te halen. Deze grenswaarden worden doorgegeven aan de kopieeractiviteit.
- Maak een kopieeractiviteit waarmee rijen uit de brongegevensopslag worden gekopieerd met een waarde uit de grenswaardekolom die groter is dan de oude grenswaarde en kleiner dan de nieuwe grenswaarde. Vervolgens worden de deltagegevens uit de brongegevensopslag als een nieuw bestand gekopieerd naar een Blob-opslag.
- Maak een opgeslagen-procedureactiviteit waarmee de grenswaarde wordt bijgewerkt voor de pijplijn die de volgende keer wordt uitgevoerd.
Als u geen Azure-abonnement hebt, maakt u een gratis account voordat u begint.
Vereisten
- Azure SQL-database. U gebruikt de database als de brongegevensopslag. Als u geen database in Azure SQL Database hebt, raadpleegt u het artikel Een database in Azure SQL Database maken om er een te maken.
- Azure Storage. U gebruikt de Blob-opslag als de sinkgegevensopslag. Als u geen opslagaccount hebt, raadpleegt u het artikel Een opslagaccount maken om een account te maken. Maak een container met de naam adftutorial.
Een gegevensbrontabel maken in uw SQL-database
Open SQL Server Management Studio. Klik in Server Explorer met de rechtermuisknop op de database en kies Nieuwe query.
Voer de volgende SQL-opdracht uit voor de SQL-database om een tabel met de naam
data_source_tablete maken als de gegevensbronopslag:create table data_source_table ( PersonID int, Name varchar(255), LastModifytime datetime ); 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');In deze zelfstudie gebruikt u LastModifytime als de grenswaardekolom. De gegevens in de brongegevensopslag worden weergegeven in de volgende tabel:
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
Nog een tabel in uw SQL-database maken om de bovengrenswaarde op te slaan
Voer de volgende SQL-opdracht uit op de SQL-database om een tabel met de naam
watermarktablete maken om de grenswaarde op te slaan:create table watermarktable ( TableName varchar(255), WatermarkValue datetime, );Stel de standaardwaarde van de bovengrens in met de tabelnaam van de brongegevensopslag. In deze zelfstudie is de tabelnaam data_source_table.
INSERT INTO watermarktable VALUES ('data_source_table','1/1/2010 12:00:00 AM')Controleer de gegevens in de tabel
watermarktable.Select * from watermarktableUitvoer:
TableName | WatermarkValue ---------- | -------------- data_source_table | 2010-01-01 00:00:00.000
Een opgeslagen procedure maken in uw SQL-database
Voer de volgende opdracht uit om een opgeslagen procedure te maken in uw SQL-database:
CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS
BEGIN
UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName
END
Een data factory maken
Start de webbrowser Microsoft Edge of Google Chrome. Op dit moment wordt de Data Factory-gebruikersinterface alleen ondersteund in de webbrowsers Microsoft Edge en Google Chrome.
Selecteer in het linkermenu Een resource maken>Integratie>Data Factory:
Voer op de pagina Nieuwe gegevensfactoryADFTutorialBulkCopyDF in als de naam.
De naam van de Azure-gegevensfactory moet wereldwijd uniek zijn. Als u een rood uitroepteken ziet met het volgende foutbericht, wijzigt u de naam van de gegevensfactory (bijvoorbeeld uwnaamADFIncCopyTutorialDF) en probeert u het opnieuw. Zie het artikel Data factory - Naamgevingsregels voor meer informatie over naamgevingsregels voor Data Factory-artefacten.
Data factory-naam 'ADFIncCopyTutorialDF' is niet beschikbaar
Selecteer het Azure-abonnement waarin u de gegevensfactory wilt maken.
Voer een van de volgende stappen uit voor de Resourcegroep:
Selecteer Bestaande gebruiken en selecteer een bestaande resourcegroep in de vervolgkeuzelijst.
Selecteer Nieuwe maken en voer de naam van een resourcegroep in.
Zie Resourcegroepen gebruiken om Azure-resources te beheren voor meer informatie.
Selecteer V2 als de versie.
Selecteer de locatie voor de gegevensfactory. In de vervolgkeuzelijst worden alleen ondersteunde locaties weergegeven. De gegevensarchieven (Azure Storage, Azure SQL Database, Azure SQL Managed Instance enzovoort) en rekenprocessen (HDInsight enzovoort) die worden gebruikt in data factory, kunnen zich in andere regio's bevinden.
Klik op Create.
Wanneer het maken is voltooid, ziet u de pagina Data Factory zoals in de afbeelding wordt weergegeven.
Selecteer Openen op de tegel Azure Data Factory Studio openen om de Gebruikersinterface (UI) van Azure Data Factory op een afzonderlijk tabblad te starten.
Een pipeline maken
In deze zelfstudie maakt u een pijplijn met twee opzoekactiviteiten, één kopieeractiviteit en één opgeslagen procedureactiviteit als keten in één pijplijn.
Klik op de startpagina van de Data Factory-gebruikersinterface op de tegel Orchestrate .
Geef op het tabblad Algemeen bij EigenschappenIncrementalCopyPipeline op als Naam. Vouw het paneel vervolgens samen door in de rechterbovenhoek op het pictogram Eigenschappen te klikken.
We gaan de eerste opzoekactiviteit gebruiken om de oudste grenswaarde op te halen. Vouw in de Activiteiten-werkset de optie Algemeen uit. Gebruik vervolgens slepen-en-neerzetten om de opzoekactiviteit te verplaatsen naar het ontwerpoppervlak voor pijplijnen. Wijzig de naam van de activiteit in LookupOldWaterMarkActivity.
Ga naar het tabblad Instellingen en klik op + Nieuw voor Brongegevensset. In deze stap maakt u een gegevensset die de gegevens in de grenswaardetabel vertegenwoordigt. Deze tabel bevat de oude grenswaarde die is gebruikt in de vorige kopieerbewerking.
Selecteer in het venster Nieuwe gegevensset de optie Azure SQL Database en klik op Doorgaan. Er wordt nu een nieuw venster geopend voor de gegevensset.
Voer in het venster Eigenschappen instellen voor de gegevensset WatermarkDataset in als Naam.
Selecteer Nieuw voor Gekoppelde service en voer de volgende stappen uit:
Voer AzureSqlDatabaseLinkedService in als Naam.
Selecteer uw server als Servernaam.
Selecteer de naam van uw database in de vervolgkeuzelijst bij Databasenaam.
Voer uw gebruikersnaam en wachtwoord in.
Als u de verbinding met uw SQL-database wilt testen, klikt u op Verbinding testen.
Klik op Voltooien.
Controleer of AzureSqlDatabaseLinkedService is geselecteerd als Gekoppelde service.
Selecteer Voltooien.
Selecteer op het tabblad Verbinding[dbo].[watermarktable] voor Tabel. Klik op Gegevens vooraf bekijken om een voorbeeld van de gegevens in de tabel te bekijken.
Ga naar de pijplijneditor door op het pijplijntabblad bovenaan te klikken of door in de structuurweergave aan de linkerkant op de naam van de pijplijn te klikken. Bevestig in het venster Eigenschappen voor de opzoekactiviteit dat WatermarkDataset is geselecteerd in het veld Brongegevensset.
Vouw in de Activiteiten-werkset de optie Algemeen uit. Gebruik vervolgens slepen-en-neerzetten om nog een opzoekactiviteit te verplaatsen naar het ontwerpoppervlak voor pijplijnen. Stel de naam op het tabblad Algemeen in het venster Eigenschappen in op LookupNewWaterMarkActivity. Met deze opzoekactiviteit wordt de nieuwe grenswaarde opgehaald uit de tabel met de brongegevens die moeten worden gekopieerd naar de bestemming.
Ga in het venster Eigenschappen voor de tweede opzoekactiviteit naar het tabblad Instellingen en klik op Nieuw. U maakt een gegevensset om te verwijzen naar de brontabel met de nieuwe grenswaarde (maximumwaarde van LastModifyTime).
Selecteer in het venster Nieuwe gegevensset de optie Azure SQL Database en klik op Doorgaan.
Voer in het venster Eigenschappen instellenSourceDataset in als Naam. Selecteer AzureSqlDatabaseLinkedService als Gekoppelde service.
Selecteer [dbo].[data_source_table] als Tabel. Verderop in de zelfstudie geeft u een query op voor deze gegevensset. De query heeft voorrang op de tabel die u in deze stap opgeeft.
Selecteer Voltooien.
Ga naar de pijplijneditor door op het pijplijntabblad bovenaan te klikken of door in de structuurweergave aan de linkerkant op de naam van de pijplijn te klikken. Bevestig in het venster Eigenschappen voor de opzoekactiviteit dat SourceDataset is geselecteerd in het veld Brongegevensset.
Selecteer Query in het veld Query gebruiken en voer de volgende query in: u selecteert alleen de maximumwaarde van LastModifytime uit data_source_table. Zorg ervoor dat u ook Alleen eerste rij hebt geselecteerd.
select MAX(LastModifytime) as NewWatermarkvalue from data_source_table
Vouw in de Activiteiten-werkset de optie Verplaatsen en transformeren uit. Gebruik vervolgens slepen-en-neerzetten om de kopieeractiviteit uit de Activiteiten-werkset te verplaatsen, en stel de naam in op IncrementalCopyActivity.
Verbind beide opzoekactiviteiten met de kopieeractiviteit door de groene knop die is gekoppeld aan de opzoekactiviteiten, naar de kopieeractiviteit te slepen. Laat de muisknop los als u ziet dat de randkleur van de kopieeractiviteit is gewijzigd in blauw.
Selecteer de kopieeractiviteit en controleer of de eigenschappen voor de activiteit worden weergegeven in het venster Eigenschappen.
Ga naar Source in het venster Properties en voer de volgende stappen uit:
Selecteer SourceDataset in het veld Brongegevensset.
Selecteer Query in het veld Query gebruiken.
Voer de volgende SQL-query in het veld Query in.
select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'
Ga naar het tabblad Sink en klik op + Nieuw voor het veld Sink-gegevensset.
In deze zelfstudie is het sink-gegevensexemplaar van het type Azure Blob-opslag. Selecteer daarom Azure Blob-opslag en klik in het venster Nieuwe gegevensset op Doorgaan.
Selecteer in het venster Indeling selecteren de indeling van uw gegevens en klik op Doorgaan.
Voer in het venster Eigenschappen instellenSinkDataset in als Naam. Selecteer + Nieuw bij Gekoppelde service. In deze stap maakt u een verbinding (gekoppelde service) voor uw Azure Blob-opslag.
Voer de volgende stappen uit in het venster Nieuwe gekoppelde service (Azure Blob Storage):
- Voer AzureStorageLinkedService in als Naam.
- Selecteer uw Azure Storage-account bij Storage account name.
- Test de verbinding en klik vervolgens op Voltooien.
Controleer of in het venster Eigenschappen instellenAzureStorageLinkedService is geselecteerd bij Gekoppelde service. Selecteer vervolgens Voltooien.
Ga naar het tabblad Verbinding van SinkDataset en voer de volgende stappen uit:
- Voer adftutorial/incrementalcopy in in het veld Bestandspad. adftutorial is de naam van de blob-container en incrementalcopy is de naam van de map. In dit fragment wordt ervan uitgegaan dat u een blobcontainer hebt met de naam adftutorial in uw Blob-opslag. Maak de container als deze bestaat niet of stel deze in op de naam van een bestaande container. Als de uitvoermap incrementalcopy niet bestaat, wordt deze automatisch gemaakt in Azure Data Factory. U kunt ook de knop Bladeren voor het bestandspad gebruiken om naar een map in een blob-container te navigeren.
- Voor het gedeelte Bestand van het veld Bestandspad selecteert u Dynamische inhoud toevoegen [Alt+P] en typt u vervolgens
@CONCAT('Incremental-', pipeline().RunId, '.txt')in het venster dat wordt geopend. Selecteer vervolgens Voltooien. De bestandsnaam wordt dynamisch gegenereerd met behulp van de expressie. Elke pijplijnuitvoering heeft een unieke id. De kopieeractiviteit gebruikt de run-id om de bestandsnaam te genereren.
Ga naar de pijplijneditor door op het pijplijntabblad bovenaan te klikken of door in de structuurweergave aan de linkerkant op de naam van de pijplijn te klikken.
Vouw in de Activiteiten-werkset de optie Algemeen uit. Gebruik vervolgens slepen-en-neerzetten om de opgeslagen-procedureactiviteit uit de Activiteiten-werkset te verplaatsen naar het ontwerpoppervlak voor pijplijnen. Verbind de groene uitvoer (geslaagd) van de kopieeractiviteit met de opgeslagen-procedureactiviteit.
Selecteer Opgeslagen procedureactiviteit in de pijplijnontwerper en verander de naam ervan in StoredProceduretoWriteWatermarkActivity.
Ga naar het tabblad SQL-account en selecteer AzureSqlDatabaseLinkedService als Gekoppelde service.
Open het tabblad Stored Procedure en voer de volgende stappen uit:
Selecteer usp_write_watermark als naam van de opgeslagen procedure.
Als u waarden wilt opgeven voor de opgeslagen-procedureparameters, klikt u op Importparameter en voert u de volgende waarden voor de parameters in:
Name Type Weergegeven als LastModifiedtime Datum en tijd @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue} TableName String @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName} 
Klik in de werkbalk op Valideren om de instellingen voor de pijplijn te valideren. Controleer of er geen validatiefouten zijn. Als u het venster Pijplijnvalidatierapport wilt sluiten, klikt u op >>.
Publiceer entiteiten (gekoppelde services, gegevenssets en pijplijnen) naar de Azure Data Factory-service door op de knop Alles publiceren te klikken. Wacht tot u een bericht ziet waarin staat dat het publiceren is voltooid.
Een pijplijnuitvoering activeren
Klik in de werkbalk op Trigger toevoegen en klik op Nu activeren.
Selecteer in het venster Pijplijnuitvoering de optie Voltooien.
De pijplijnuitvoering controleren.
Ga naar het tabblad Controleren aan de linkerkant. U ziet de status van de pijplijnuitvoering die is geactiveerd met een handmatige trigger. U kunt via koppelingen in de kolom NAAM PIJPLIJN uitvoeringsdetails bekijken en de pijplijn opnieuw uitvoeren.
Selecteer de koppeling in de kolom NAAM PIJPLIJN om de uitvoering van activiteiten te zien die zijn gekoppeld aan de pijplijnuitvoering. Selecteer de koppeling Details (pictogram van een bril) in de kolom NAAM ACTIVITEIT om details van de uitvoeringen van een activiteit te zien. Selecteer Alle pijplijnuitvoeringen bovenaan om terug te gaan naar de weergave Pijplijnuitvoeringen. Selecteer Vernieuwen om de weergave te vernieuwen.
De resultaten bekijken
Maak verbinding met uw Azure Storage-account met behulp van hulpprogramma's zoals Azure Storage Explorer. Controleer of er een uitvoerbestand is gemaakt in de map incrementalcopy in de container adftutorial.
Als u het uitvoerbestand opent, ziet u dat alle gegevens uit data_source_table zijn gekopieerd naar het blob-bestand.
1,aaaa,2017-09-01 00:56:00.0000000 2,bbbb,2017-09-02 05:23:00.0000000 3,cccc,2017-09-03 02:36:00.0000000 4,dddd,2017-09-04 03:21:00.0000000 5,eeee,2017-09-05 08:06:00.0000000Controleer de laatste waarde van
watermarktable. U ziet dat de grenswaarde is bijgewerkt.Select * from watermarktableDit is de uitvoer:
| TableName | WatermarkValue | | --------- | -------------- | | data_source_table | 2017-09-05 8:06:00.000 |
Meer gegevens toevoegen aan de bron
Voeg nieuwe gegevens in uw database (brongegevensopslag) in.
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')
De bijgewerkte gegevens in uw database zijn:
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
Een andere pijplijnuitvoering activeren
Ga naar het tabblad Bewerken . Klik op de pijplijn in de structuurweergave als deze niet is geopend in de ontwerpfunctie.
Klik in de werkbalk op Trigger toevoegen en klik op Nu activeren.
Controleer de tweede pijplijnuitvoering.
Ga naar het tabblad Controleren aan de linkerkant. U ziet de status van de pijplijnuitvoering die is geactiveerd met een handmatige trigger. U kunt via koppelingen in de kolom NAAM PIJPLIJN details van activiteiten bekijken en de pijplijn opnieuw uitvoeren.
Selecteer de koppeling in de kolom NAAM PIJPLIJN om de uitvoering van activiteiten te zien die zijn gekoppeld aan de pijplijnuitvoering. Selecteer de koppeling Details (pictogram van een bril) in de kolom NAAM ACTIVITEIT om details van de uitvoeringen van een activiteit te zien. Selecteer Alle pijplijnuitvoeringen bovenaan om terug te gaan naar de weergave Pijplijnuitvoeringen. Selecteer Vernieuwen om de weergave te vernieuwen.
De tweede uitvoer controleren
In de Blob-opslag ziet u dat een ander bestand is gemaakt. In deze zelfstudie is de nieuwe bestandsnaam
Incremental-<GUID>.txt. Als u dit bestand opent, ziet u twee rijen met records.6,newdata,2017-09-06 02:23:00.0000000 7,newdata,2017-09-07 09:01:00.0000000Controleer de laatste waarde van
watermarktable. U ziet dat de grenswaarde opnieuw is bijgewerkt.Select * from watermarktableVoorbeelduitvoer:
| TableName | WatermarkValue | | --------- | -------------- | | data_source_table | 2017-09-07 09:01:00.000 |
Gerelateerde inhoud
In deze zelfstudie hebt u de volgende stappen uitgevoerd:
- Bereid de gegevensopslag voor om de grenswaarde in op te slaan.
- Een data factory maken.
- Maak gekoppelde services.
- Maak bron-, sink- en grenswaardegegevenssets.
- Een pipeline maken.
- Voer de pijplijn uit.
- Controleer de pijplijnuitvoering.
- Resultaten controleren
- Voeg meer gegevens toe aan de bron.
- Voer de pijplijn opnieuw uit.
- Controleer de tweede pijplijnuitvoering.
- Bekijk de resultaten van de tweede uitvoering.
In deze zelfstudie heeft de pijplijn gegevens uit één tabel in een SQL-database naar een Blob-opslag gekopieerd. Ga door naar de volgende zelfstudie voor meer informatie over het kopiëren van gegevens uit meerdere tabellen in een SQL Server-database naar SQL Database.