Kopiowanie różnicowe z bazy danych z tabelą kontrolną

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym artykule opisano szablon, który jest dostępny do przyrostowego ładowania nowych lub zaktualizowanych wierszy z tabeli bazy danych do Azure przy użyciu zewnętrznej tabeli sterowania, która przechowuje wartość wysokiego limitu.

Ten szablon wymaga, aby schemat źródłowej bazy danych zawierał kolumnę sygnatury czasowej lub klucz przyrostowy w celu zidentyfikowania nowych lub zaktualizowanych wierszy.

Informacje o tym szablonie rozwiązania

Ten szablon najpierw pobiera starą wartość limitu i porównuje go z bieżącą wartością limitu. Następnie kopiuje tylko zmiany w źródłowej bazie danych na podstawie porównania dwóch wartości znacznika. Na koniec zapisuje nową wartość wysokiego limitu w zewnętrznej tabeli sterowania na potrzeby ładowania danych różnicowych następnym razem.

Szablon zawiera cztery działania:

• Funkcja wyszukiwania pobiera starą wartość najwyższą, która jest przechowywana w zewnętrznej tabeli kontrolnej.
• Inne działanie Lookup pobiera bieżącą wartość limitu górnego ze źródłowej bazy danych.
• Kopiuje tylko zmiany ze źródłowej bazy danych do docelowego repozytorium. Zapytanie identyfikujące zmiany w źródłowej bazie danych jest podobne do "SELECT * FROM Data_Source_Table WHERE TIMESTAMP_Column > "last high-watermark" i TIMESTAMP_Column <= "current high-watermark".
• SqlServerStoredProcedure zapisuje bieżącą wartość znacznika granicznego w zewnętrznej tabeli sterowania dla kopiowania różnicowego przy następnym użyciu.

Szablon definiuje następujące parametry:

• Data_Source_Table_Name to tabela w źródłowej bazie danych, z której chcesz załadować dane.
• Data_Source_WaterMarkColumn to nazwa kolumny w tabeli źródłowej używanej do identyfikowania nowych lub zaktualizowanych wierszy. Typ tej kolumny to zazwyczaj data/godzina, INT lub podobne.
• Data_Destination_Container jest ścieżką główną miejsca, w którym dane są kopiowane do magazynu docelowego.
• Data_Destination_Directory to ścieżka katalogu znajdująca się pod katalogiem głównym miejsca docelowego, gdzie dane są kopiowane do magazynu docelowego.
• Data_Destination_Table_Name to miejsce, w którym dane są kopiowane do docelowego magazynu danych (gdy wybrano "Azure Synapse Analytics" jako miejsce docelowe).
• Data_Destination_Folder_Path to miejsce, do którego kopiowane są dane w magazynie docelowym (gdy jako miejsce docelowe danych wybrano "System plików" lub "Azure Data Lake Storage Gen1").
• Control_Table_Table_Name to zewnętrzna tabela kontrolna, która przechowuje wartość punktu odniesienia.
• Control_Table_Column_Name to kolumna w tabeli kontroli zewnętrznej, która przechowuje wartość wysokiego limitu.

Jak używać tego szablonu rozwiązania

1. Zapoznaj się z tabelą źródłową, którą chcesz załadować, i zdefiniuj kolumnę punktu odniesienia, która może służyć do identyfikowania nowych lub zaktualizowanych wierszy. Typ tej kolumny może być data/godzina, INT lub podobny. Wartość tej kolumny zwiększa się w miarę dodawania nowych wierszy. Z poniższej przykładowej tabeli źródłowej (data_source_table) możemy użyć kolumny LastModifytime jako kolumny limitu górnego.

   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           fffffff         2017-09-06 02:23:00.000
   7           gggg            2017-09-07 09:01:00.000
   8           hhhh            2017-09-08 09:01:00.000
   9           iiiiiiiii       2017-09-09 09:01:00.000
   

2. Utwórz tabelę kontrolną w SQL Server lub Azure SQL Database, aby przechowywać wartość wskaźnika szczytowego dla ładowania danych różnicowych. W poniższym przykładzie nazwa tabeli kontrolnej to watermarktable. W tej tabeli WatermarkValue to kolumna, w której przechowywana jest wartość limitu górnego, a jej typ to datetime.

   create table watermarktable
   (
   WatermarkValue datetime,
   );
   INSERT INTO watermarktable
   VALUES ('1/1/2010 12:00:00 AM')
   

3. Utwórz procedurę składowaną w tym samym wystąpieniu SQL Server lub Azure SQL Database, które użyto do utworzenia tabeli kontrolnej. Procedura składowana służy do zapisywania nowej wartości górnej limitu w zewnętrznej tabeli sterowania na potrzeby ładowania danych różnicowych przy następnym ładowaniu.

   CREATE PROCEDURE update_watermark @LastModifiedtime datetime
   AS
   
   BEGIN
   
       UPDATE watermarktable
       SET [WatermarkValue] = @LastModifiedtime 
   
   END
   

4. Przejdź do szablonu kopii różnicowej z bazy danych. Utwórz nowe połączenie ze źródłową bazą danych, z której chcesz skopiować dane.

   

5. Utwórz nowe połączenie z docelowym magazynem danych, do którego chcesz skopiować dane.

   

6. Utwórz Nowe połączenie z tabelą kontroli zewnętrznej i procedurą składowaną utworzoną w krokach 2 i 3.

   

7. Wybierz Użyj tego szablonu.

8. Widoczny jest dostępny potok, na przykładzie poniżej:

   

9. Wybierz Procedurę składowaną. W polu Nazwa procedury składowanej wybierz pozycję [dbo].[update_watermark]. Wybierz pozycję Importuj parametr, a następnie wybierz pozycję Dodaj zawartość dynamiczną.

   

10. Napisz zawartość @{activity('LookupCurrentWaterMark').output.firstRow.NewWatermarkValue}, a następnie wybierz Zakończ.

    

11. Wybierz pozycję Debuguj, wprowadź parametry, a następnie wybierz pozycję Zakończ.

    

12. Zostaną wyświetlone wyniki podobne do poniższego przykładu:

    

13. Możesz utworzyć nowe wiersze w tabeli źródłowej. Oto przykładowy język SQL do tworzenia nowych wierszy:

    INSERT INTO data_source_table
    VALUES (10, 'newdata','9/10/2017 2:23:00 AM')
    
    INSERT INTO data_source_table
    VALUES (11, 'newdata','9/11/2017 9:01:00 AM')
    

14. Aby ponownie uruchomić potok, wybierz Debuguj, wprowadź Parametry, a następnie wybierz Zakończ.

    Zobaczysz, że do miejsca docelowego zostały skopiowane tylko nowe wiersze.

15. (Opcjonalnie:) Jeśli wybierzesz Azure Synapse Analytics jako miejsce docelowe danych, musisz również podać połączenie z usługą Azure Blob Storage na potrzeby etapowania, co jest wymagane przez funkcję Azure Synapse Analytics Polybase. Szablon utworzy dla Ciebie ścieżkę kontenera. Po uruchomieniu potoku sprawdź, czy kontener został utworzony w magazynie Blob.

    

Powiązana zawartość

• Kopia zbiorcza z bazy danych przy użyciu tabeli sterującej w Azure Data Factory
• Kopiowanie plików z wielu kontenerów z użyciem Azure Data Factory