Przyrostowe ładowanie danych z usługi Azure SQL Database do usługi Azure Blob Storage przy użyciu witryny Azure Portal

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym samouczku utworzysz Azure Data Factory z potokiem, który ładuje dane delta z tabeli w Azure SQL Database do Azure Blob Storage.

Ten samouczek obejmuje następujące procedury:

Omówienie

Diagram ogólny rozwiązania wygląda następująco:

Poniżej przedstawiono ważne czynności związane z tworzeniem tego rozwiązania:

1. Wybierz kolumnę limitu. Wybierz jedną kolumnę w magazynie danych źródłowych, która może służyć do filtrowania nowych lub zaktualizowanych rekordów dla każdego uruchomienia. Zazwyczaj dane w tej wybranej kolumnie (na przykład last_modify_time lub ID) rosną wraz z tworzeniem i aktualizacją wierszy. Maksymalna wartość w tej kolumnie jest używana jako limit.

2. Przygotuj magazyn danych do przechowywania wartości limitu. W tym samouczku wartość limitu jest przechowywana w bazie danych SQL.

3. Utwórz potok z następującym przepływem pracy:

   Pipeline w tym rozwiązaniu obejmuje następujące działania:

   • Utwórz dwa działania wyszukiwania. Użyj pierwszego zadania Lookup, aby pobrać ostatnią wartość watermark. Użyj drugiego działania Lookup do pobrania nowej wartości limitu. Te wartości limitu są przekazywane do działania Copy.
   • Utwórz działanie Copy, które kopiuje wiersze z magazynu danych źródłowych o wartości kolumny limitu większej niż poprzednia wartość limitu i mniejszej niż nowa wartość limitu. Następnie kopiuje dane różnicowe ze źródłowego magazynu danych do usługi Blob Storage jako nowy plik.
   • Utwórz działanie StoredProcedure, które aktualizuje wartość znacznika dla potoku, który uruchomi się następnym razem.

Jeśli nie masz subskrypcji platformy Azure, przed rozpoczęciem utwórz bezpłatne konto.

Wymagania wstępne

• usługi Azure SQL Database. Baza danych jest używana jako źródłowy magazyn danych. Jeśli nie masz bazy danych w usłudze Azure SQL Database, zobacz Tworzenie bazy danych w usłudze Azure SQL Database , aby uzyskać instrukcje tworzenia bazy danych.
• Usługa Azure Storage. Używasz magazynu obiektów blob jako magazynu danych ujścia. Jeśli nie masz konta magazynu, utwórz je, wykonując czynności przedstawione w artykule Tworzenie konta magazynu. Utwórz kontener o nazwie adftutorial.

Tworzenie tabeli danych źródłowych w bazie danych SQL

1. Otwórz SQL Server Management Studio. W Eksploratorze serwera kliknij prawym przyciskiem myszy bazę danych, a następnie wybierz pozycję Nowe zapytanie.

2. Uruchom następujące polecenie SQL w bazie danych SQL, aby utworzyć tabelę o nazwie data_source_table jako magazyn źródła danych.

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

   W tym samouczku użyjesz kolumny LastModifytime jako kolumny limitu. Dane w źródłowym magazynie danych zostały przedstawione w poniższej tabeli:

   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
   

Tworzenie innej tabeli w bazie danych SQL do przechowywania wartości górnego limitu

1. Uruchom następujące polecenie SQL względem bazy danych SQL, aby utworzyć tabelę o nazwie watermarktable w celu przechowywania wartości znacznika.

   create table watermarktable
   (
   
   TableName varchar(255),
   WatermarkValue datetime,
   );
   

2. Ustaw domyślną wartość górnego limitu, używając nazwy tabeli w źródłowym magazynie danych. W tym samouczku nazwa tabeli to data_source_table.

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

3. Sprawdź dane w tabeli watermarktable.

   Select * from watermarktable
   

   Wyjście:

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

Tworzenie procedur składowanych w bazie danych SQL

Uruchom następujące polecenie, aby utworzyć procedurę składowaną w bazie danych SQL:

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

BEGIN

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

END

Tworzenie fabryki danych

1. Uruchom przeglądarkę internetową Microsoft Edge lub Google Chrome. Obecnie interfejs użytkownika usługi Data Factory jest obsługiwany tylko przez przeglądarki internetowe Microsoft Edge i Google Chrome.

2. W górnym menu wybierz pozycję Utwórz zasób>Analytics>Data Factory :

   

3. Na stronie Nowa fabryka danych, wprowadź ADFIncCopyTutorialDF jako nazwę.

   Nazwa usługi Azure Data Factory musi być globalnie unikatowa. Jeśli pojawi się czerwony wykrzyknik z poniższym błędem, zmień nazwę fabryki danych (np. twojanazwaADFIncCopyTutorialDF) i spróbuj utworzyć ją ponownie. Artykuł Data Factory — Naming Rules (Usługa Data Factory — reguły nazewnictwa) zawiera reguły nazewnictwa artefaktów usługi Data Factory.

   Nazwa fabryki danych "ADFIncCopyTutorialDF" jest niedostępna

4. Wybierz subskrypcję Azure, w której chcesz utworzyć fabrykę danych.

5. Dla opcji Grupa zasobów wykonaj jedną z następujących czynności:

   • Wybierz pozycję Użyj istniejącej, a następnie wybierz istniejącą grupę zasobów z listy rozwijanej.

   • Wybierz pozycję Utwórz nową, a następnie wprowadź nazwę grupy zasobów.

     Informacje na temat grup zasobów znajdują się w artykule Using resource groups to manage your Azure resources (Używanie grup zasobów do zarządzania zasobami platformy Azure).

6. Wybierz V2 dla wersji.

7. Wybierz lokalizację dla fabryki danych. Na liście rozwijanej są wyświetlane tylko obsługiwane lokalizacje. Magazyny danych (Azure Storage, Azure SQL Database, Azure SQL Managed Instance itd.) i obliczenia (HDInsight itp.) używane przez fabrykę danych mogą znajdować się w innych regionach.

8. Kliknij pozycję Utwórz.

9. Po zakończeniu tworzenia zostanie wyświetlona strona Fabryka danych, jak pokazano na poniższej ilustracji.

   

10. Wybierz opcję Otwórz na kafelku Open Azure Data Factory Studio, aby uruchomić interfejs użytkownika usługi Azure Data Factory w osobnej karcie.

Tworzenie pipeline'u

W tym samouczku utworzysz potok z dwoma aktywnościami Lookup, jedną aktywnością Copy i jedną aktywnością StoredProcedure, wszystkie połączone w jednym potoku.

1. Na stronie głównej interfejsu użytkownika usługi Data Factory kliknij kafelek Orkiestracja .

   

2. W panelu Ogólne w obszarze Właściwości określ IncrementalCopyPipeline dla Nazwa. Następnie zwiń panel, klikając ikonę Właściwości w prawym górnym rogu.

3. Dodajmy pierwszą operację przeszukiwania w celu pobrania wartości starego znacznika. W przyborniku Działania rozwiń pozycję Ogólne, a następnie przeciągnij działanie Lookup (Wyszukiwanie) i upuść je na powierzchni projektanta potoku. Zmień nazwę działania na LookupOldWaterMarkActivity.

   

4. Przejdź do karty Ustawienia, a następnie kliknij pozycję +Nowy dla pozycji Źródłowy zestaw danych. W tym kroku utworzysz zestaw danych reprezentujący dane w tabeli watermarktable. Ta tabela zawiera stary znak wodny, który był używany w poprzedniej operacji kopiowania.

5. W oknie Nowy zestaw danych wybierz pozycję Azure SQL Database, a następnie kliknij przycisk Kontynuuj. Zostanie otwarte nowe okno dla zestawu danych.

6. W oknie Ustawianie właściwości zestawu danych wprowadź wartość WatermarkDataset w polu Nazwa.

7. W obszarze Połączona usługa wybierz pozycję Nowa, a następnie wykonaj następujące czynności:

   1. Wprowadź wartość AzureSqlDatabaseLinkedService w polu Nazwa.

   2. Wybierz serwer w polu Nazwa serwera.

   3. Wybierz swoją nazwę bazy danych z listy rozwijanej.

   4. Wprowadź nazwę użytkownika i hasło.

   5. Aby przetestować połączenie z bazą danych SQL, kliknij pozycję Testuj połączenie.

   6. Kliknij przycisk Zakończ.

   7. Upewnij się, że dla połączonej usługi wybrano AzureSqlDatabaseLinkedService.

      

   8. Wybierz Zakończ.

8. Na karcie Połączenie wybierz pozycję [dbo].[watermarktable] dla tabeli. Jeśli chcesz wyświetlić podgląd danych w tabeli, kliknij przycisk Podgląd danych.

   

9. Przejdź do edytora potoku, klikając kartę potoku u góry lub klikając nazwę potoku w widoku drzewa po lewej stronie. W oknie właściwości działania Lookup (Wyszukiwanie) upewnij się, że dla pola Zestaw danych źródłowych wybrano WatermarkDataset.

10. W przyborniku Działania rozwiń pozycję Ogólne, a następnie przeciągnij i upuść kolejną czynność Lookup na powierzchnię projektanta potoku i ustaw nazwę na LookupNewWaterMarkActivity w karcie Ogólne okna właściwości. To działanie wyszukiwania pobiera nową wartość limitu z tabeli zawierającej dane źródłowe do skopiowania do miejsca docelowego.

11. W oknie właściwości dla drugiego działania Lookup (Wyszukiwania) przejdź do karty Ustawienia, a następnie kliknij pozycję Nowy. Utworzysz zestaw danych wskazujący tabelę źródłową zawierającą nową wartość limitu (wartość maksymalna elementu LastModifyTime).

12. W oknie Nowy zestaw danych wybierz pozycję Azure SQL Database, a następnie kliknij przycisk Kontynuuj.

13. W oknie Ustaw właściwości wprowadź SourceDataset dla Nazwa. Wybierz wartość AzureSqlDatabaseLinkedService w polu Połączona usługa.

14. Wybierz [dbo].[data_source_table] dla tabeli. Wprowadzasz zapytanie dla tego zestawu danych później w tym samouczku. Zapytanie ma pierwszeństwo przed tabelą określaną w tym kroku.

15. Wybierz Zakończ.

16. Przejdź do edytora potoku, klikając kartę potoku u góry lub klikając nazwę potoku w widoku drzewa po lewej stronie. W oknie dialogowym właściwości działania Lookup upewnij się, że SourceDataset jest wybrany w polu Source Dataset.

17. Wybierz wartość Zapytanie w polu Użyj zapytania, a następnie wprowadź następujące zapytanie: wybierasz tylko maksymalną wartość LastModifytime z tabeli data_source_table. Upewnij się, że masz również zaznaczone Tylko pierwszy wiersz.

    select MAX(LastModifytime) as NewWatermarkvalue from data_source_table
    

    

18. W przyborniku Działania rozwiń pozycję Przenieś i przekształć, a następnie przeciągnij działanie Kopiuj z przybornika Działania i ustaw nazwę na IncrementalCopyActivity.

19. Połącz oba działania wyszukiwania z działaniem kopiowania, przeciągając do działania kopiowania zielony przycisk dołączony do działania wyszukiwania. Po zmianie koloru obramowania działania kopiowania na niebieski zwolnij przycisk myszy.

    

20. Wybierz działanie kopiowania i upewnij się, że w oknie Właściwości widać właściwości działania.

21. Przejdź do karty Źródło w oknie Właściwości i wykonaj następujące czynności:

    1. Wybierz pozycję SourceDataset dla pola Zestaw danych będący źródłem.

    2. Wybierz pozycję Zapytanie dla pola Użyj zapytania.

    3. W polu Zapytanie wprowadź następujące zapytanie SQL.

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

       

22. Przejdź do karty Ujście i kliknij pozycję +Nowy dla pola Zestaw danych ujścia.

23. W tym samouczku docelowy magazyn danych jest typu Azure Blob Storage. W związku z tym wybierz pozycję Azure Blob Storage i kliknij przycisk Kontynuuj w oknie Nowy zestaw danych .

24. W oknie Wybieranie formatu wybierz typ formatu danych, a następnie kliknij przycisk Kontynuuj.

25. W oknie Ustawianie właściwości wprowadź SinkDataset w polu Nazwa. W obszarze Połączona usługa wybierz pozycję + Nowa. W tym kroku tworzysz połączenie (usługę połączoną) z magazynem Azure Blob Storage.

26. W oknie Nowa połączona usługa (Azure Blob Storage) wykonaj następujące czynności:

    1. Wprowadź wartość AzureStorageLinkedService w polu Nazwa.
    2. W polu Nazwa konta magazynu wybierz konto usługi Azure Storage.
    3. Przetestuj połączenie, a następnie kliknij przycisk Zakończ.

27. W oknie Ustawianie właściwości upewnij się, że dla połączonej usługi wybrano pozycję AzureStorageLinkedService. Następnie wybierz pozycję Zakończ.

28. Przejdź do karty Połączenie zestawu SinkDataset i wykonaj następujące czynności:

    1. W polu ścieżka pliku wprowadź wartość adftutorial/incrementalcopy. Nazwą kontenera obiektów blob jest adftutorial, a nazwą folderu jest incrementalcopy. W tym fragmencie kodu założono, że masz kontener blob nazwany adftutorial w magazynie blobów. Utwórz ten kontener, jeśli nie istnieje, lub zastąp go nazwą istniejącego kontenera. Jeśli folder wyjściowy incrementalcopy nie istnieje, usługa Azure Data Factory automatycznie go utworzy. Można również użyć przycisku Przeglądaj w polu Ścieżka pliku, aby przejść do folderu w kontenerze obiektów blob.
    2. Dla części Plik w polu Ścieżka pliku wybierz pozycję Dodaj zawartość dynamiczną [Alt+P], a następnie w otwartym oknie wprowadź @CONCAT('Incremental-', pipeline().RunId, '.txt'). Następnie wybierz pozycję Zakończ. Nazwa pliku jest generowana dynamicznie przy użyciu wyrażenia. Każde uruchomienie potoku ma unikatowy identyfikator. Działanie kopiowania używa identyfikatora uruchomienia do wygenerowania nazwy pliku.

29. Aby przejść do edytora potoku, kliknij kartę potoku u góry ekranu lub nazwę potoku w widoku drzewa z lewej strony.

30. W przyborniku Działania rozwiń pozycję Ogólne, a następnie przeciągnij i upuść działanie Stored Procedure (Procedura składowana) z przybornika Działania na powierzchnię projektanta potoku. Połącz zielone (Powodzenie) wyjście działania kopiowania z działaniem procedury składowanej.

31. Wybierz Działanie Procedury Składowanej w projektancie potoku i zmień jego nazwę na StoredProceduretoWriteWatermarkActivity.

32. Przejdź do karty SQL Account, a następnie wybierz AzureSqlDatabaseLinkedService dla Usługa połączona.

33. Przejdź do karty Procedura składowana i wykonaj następujące czynności:

    1. W polu Nazwa procedury składowanej wybierz wartość usp_write_watermark.

    2. Aby określić wartości parametrów procedury składowanej, kliknij pozycję Importuj parametr, a następnie wprowadź następujące wartości parametrów:

       Nazwisko	Typ	Wartość
       CzasOstatniejModyfikacji	Data i Czas	@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
       NazwaTabeli	Sznurek	@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

       

34. Aby zweryfikować ustawienia przepływu, kliknij pozycję Weryfikuj na pasku narzędzi. Potwierdź, że nie ma błędów walidacji. Aby zamknąć okno Raport weryfikacji Pipeline, kliknij >>.

35. Opublikuj jednostki (usługi połączone, zestawy danych i potoki) w usłudze Azure Data Factory, wybierając przycisk Opublikuj Wszystko. Poczekaj, aż zostanie wyświetlony komunikat o pomyślnym opublikowaniu.

Wyzwalanie uruchomienia potoku

1. Kliknij pozycję Dodaj wyzwalacz na pasku narzędzi, a następnie kliknij pozycję Wyzwól teraz.

2. W oknie Uruchomienie potoku wybierz Zakończ.

Monitorowanie działania przepływu danych

1. Przejdź do karty Monitorowanie po lewej stronie. Zostanie wyświetlony stan uruchomienia potoku wyzwalanego przez wyzwalacz ręczny. Możesz użyć linków w kolumnie NAZWA POTOKU, aby wyświetlić szczegóły przebiegu i ponownie uruchomić potok.

2. Aby wyświetlić przebiegi działań powiązane z uruchomieniem potoku, wybierz link pod kolumną NAZWA POTOKU. Aby uzyskać szczegółowe informacje na temat przebiegów działań, wybierz link Szczegóły (ikona okularów) w kolumnie NAZWA DZIAŁANIA. Wybierz pozycję Wszystkie uruchomienia potoków u góry, aby wrócić do widoku Uruchomienia potoków. Aby odświeżyć widok, wybierz pozycję Odśwież.

Sprawdzanie wyników

1. Połącz się z kontem usługi Azure Storage za pomocą narzędzi, takich jak Eksplorator usługi Azure Storage. Sprawdź, czy plik wyjściowy został utworzony w folderze incrementalcopy w kontenerze adftutorial.

   

2. Otwórz plik wyjściowy i zwróć uwagę na to, że do pliku obiektu blob skopiowano wszystkie dane z tabeli data_source_table.

   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.0000000
   

3. Sprawdź najnowszą wartość z watermarktable. Zobaczysz, że wartość limitu została zaktualizowana.

   Select * from watermarktable
   

   Oto dane wyjściowe:

   | TableName | WatermarkValue |
   | --------- | -------------- |
   | data_source_table | 2017-09-05	8:06:00.000 |
   

Dodawanie większej ilości danych do źródła

Wstaw nowe dane do bazy danych (magazyn źródeł danych).

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

Zaktualizowane dane w bazie danych to:

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

Uruchom ponownie potok

1. Przejdź do karty Edycja. Kliknij pipeline w widoku drzewa, jeśli nie został otwarty w projektancie.

2. Kliknij pozycję Dodaj wyzwalacz na pasku narzędzi, a następnie kliknij pozycję Wyzwól teraz.

Monitorowanie drugiego uruchomienia potoku

1. Przejdź do karty Monitorowanie po lewej stronie. Zostanie wyświetlony stan uruchomienia potoku wyzwalanego przez wyzwalacz ręczny. Możesz użyć linków w kolumnie NAZWA POTOKU, aby wyświetlić szczegóły działania i ponownie uruchomić potok.

2. Aby wyświetlić przebiegi działań powiązane z uruchomieniem potoku, wybierz link pod kolumną NAZWA POTOKU. Aby uzyskać szczegółowe informacje na temat przebiegów działań, wybierz link Szczegóły (ikona okularów) w kolumnie NAZWA DZIAŁANIA. Wybierz pozycję Wszystkie uruchomienia potoków u góry, aby wrócić do widoku Uruchomienia potoków. Aby odświeżyć widok, wybierz pozycję Odśwież.

Weryfikowanie drugiego zestawu danych wyjściowych

1. W przechowywaniu BLOB zobaczysz, że utworzono kolejny plik. W tym samouczku nazwa nowego pliku to Incremental-<GUID>.txt. Otwórz ten plik — zobaczysz w nim dwa wiersze rekordów.

   6,newdata,2017-09-06 02:23:00.0000000
   7,newdata,2017-09-07 09:01:00.0000000    
   

2. Sprawdź najnowszą wartość z watermarktable. Zobaczysz, że wartość limitu została ponownie zaktualizowana.

   Select * from watermarktable
   

   Przykładowe dane wyjściowe:

   | TableName | WatermarkValue |
   | --------- | -------------- |
   | data_source_table | 2017-09-07 09:01:00.000 |
   

Powiązana zawartość

W ramach tego samouczka wykonano następujące procedury:

W tym samouczku proces skopiował dane z jednej tabeli w usłudze SQL Database do magazynu BLOB Storage. Przejdź do poniższego samouczka, aby dowiedzieć się, jak kopiować dane z wielu tabel w bazie danych programu SQL Server do usługi SQL Database.