Przyrostowe ładowanie danych z Azure SQL Managed Instance do Azure Storage przy użyciu przechwytywania zmian danych (CDC)

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym samouczku utworzysz fabrykę danych Azure z potokiem, który na podstawie informacji o przechwytywaniu danych zmiany (CDC) w źródłowej bazie danych Azure SQL Managed Instance ładuje dane różnicowe do magazynu obiektów blob Azure.

W tym samouczku wykonasz następujące kroki:

Omówienie

Technologia przechwytywania zmian danych, obsługiwana przez magazyny danych, takie jak Azure SQL Managed Instances (MI) i SQL Server, mogą służyć do identyfikowania zmienionych danych. W tym podręczniku opisano, jak używać Azure Data Factory z technologią śledzenia zmian danych SQL, aby przyrostowo załadować dane delta z Azure SQL Managed Instance do Azure Blob Storage. Aby uzyskać bardziej szczegółowe informacje na temat technologii przechwytywania zmian danych SQL, zobacz Zmień przechwytywanie danych w SQL Server.

Kompletny przepływ pracy

Poniżej przedstawiono typowe kompleksowe kroki przepływu pracy umożliwiające przyrostowe ładowanie danych przy użyciu technologii Change Data Capture.

Rozwiązanie na wysokim poziomie

W tym samouczku utworzysz pipeline, który wykonuje następujące operacje:

1. Utwórz operację wyszukiwania, aby zliczyć liczbę zmienionych rekordów w tabeli CDC SQL Database i przekazać je do działania typu IF.
2. Utwórz warunek if, aby sprawdzić, czy istnieją zmienione rekordy i jeśli tak, wywołaj działanie kopiowania.
3. Utwórz zadanie kopiowania do kopiowania wstawionych/zaktualizowanych/usuniętych danych z tabeli CDC do Azure Blob Storage.

Jeśli nie masz subskrypcji Azure, przed rozpoczęciem utwórz konto free.

Wymagania wstępne

• Azure SQL Managed Instance. Baza danych jest używana jako magazyn danych źródłowych. Jeśli nie masz Azure SQL Managed Instance, zobacz artykuł Tworzenie Azure SQL Database Managed Instance aby uzyskać instrukcje tworzenia.
• konto Azure Storage. Magazyn obiektów blob jest używany jako magazyn danych docelowy. Jeśli nie masz konta magazynu Azure, zobacz artykuł Utwórz konto magazynu aby uzyskać instrukcje tworzenia konta. Utwórz kontener o nazwie raw.

Tworzenie tabeli źródła danych w Azure SQL Database

1. Uruchom SQL Server Management Studio i połącz się z serwerem Azure SQL Managed Instances.

2. W Eksploratorze serwera kliknij prawym przyciskiem używaną bazę danych, a następnie wybierz pozycję Nowe zapytanie.

3. Uruchom następujące polecenie SQL względem bazy danych Azure SQL Managed Instances, aby utworzyć tabelę o nazwie customers jako magazyn źródeł danych.

   create table customers 
   (
   customer_id int, 
   first_name varchar(50), 
   last_name varchar(50), 
   email varchar(100), 
   city varchar(50), CONSTRAINT "PK_Customers" PRIMARY KEY CLUSTERED ("customer_id") 
    );
   

4. Włącz mechanizm przechwytywania zmian danych w bazie danych i tabeli źródłowej (klientów), uruchamiając następujące zapytanie SQL:

   Uwaga

   • Zastąp <nazwę schematu źródłowego> schematem Azure SQL MI zawierającym tabelę klienci.
   • Przechwytywanie zmian danych nie wykonuje żadnych czynności w ramach transakcji, które zmieniają śledzone tabele. Zamiast tego operacje wstawiania, aktualizowania i usuwania są zapisywane w dzienniku transakcji. Dane, które są zdeponowane w tabelach zmian, staną się niezarządzane, jeśli dane nie będą okresowo i systematycznie czyszczone. Aby uzyskać więcej informacji, zobacz Włączanie przechwytywania zmian danych dla bazy danych

   EXEC sys.sp_cdc_enable_db 
   
   EXEC sys.sp_cdc_enable_table
   @source_schema = 'dbo',
   @source_name = 'customers', 
   @role_name = NULL,
   @supports_net_changes = 1
   

5. Wstaw dane do tabeli customers, uruchamiając następujące polecenie:

    insert into customers 
        (customer_id, first_name, last_name, email, city) 
    values 
        (1, 'Chevy', 'Leward', 'cleward0@mapy.cz', 'Reading'),
        (2, 'Sayre', 'Ateggart', 'sateggart1@nih.gov', 'Portsmouth'),
       (3, 'Nathalia', 'Seckom', 'nseckom2@blogger.com', 'Portsmouth');
   

   Uwaga

   Żadne historyczne zmiany w tabeli nie są przechwytywane przed włączeniem przechwytywania zmian danych.

Tworzenie fabryki danych

Wykonaj kroki opisane w artykule Quickstart: Tworzenie fabryki danych przy użyciu portalu Azure, aby utworzyć fabrykę danych, jeśli jeszcze jej nie masz do pracy.

Tworzenie połączonych usług

Połączone usługi tworzy się w fabryce danych w celu połączenia magazynów danych i usług obliczeniowych z fabryką danych. W tej sekcji utworzysz usługi powiązane z twoim kontem Azure Storage i Azure SQL Managed Instance.

Utwórz połączoną usługę Azure Storage.

W tym kroku połączysz konto Azure Storage z fabryką danych.

1. Kliknij Połączenia, a następnie + Nowy.

   

2. W oknie Nowa połączona usługa wybierz Azure Blob Storage i kliknij Continue.

   

3. W oknie Nowa połączona usługa wykonaj następujące czynności:

   1. Wprowadź wartość AzureStorageLinkedService w polu Nazwa.
   2. Wybierz konto magazynu Azure dla Nazwa konta magazynu.
   3. Kliknij przycisk Zapisz.

   

Utwórz połączoną usługę Azure SQL MI Database.

W tym kroku połączysz bazę danych Azure SQL MI z fabryką danych.

1. Kliknij Połączenia, a następnie + Nowy.

2. W oknie Nowa połączona usługa wybierz Azure SQL Database Managed Instance i kliknij Continue.

3. W oknie Nowa połączona usługa wykonaj następujące czynności:

   1. Wprowadź AzureSqlMI1 w polu Nazwa.
   2. Wybierz serwer SQL w polu Nazwa serwera.
   3. Wybierz bazę danych SQL dla pola Nazwa bazy danych.
   4. W polu Nazwa użytkownika podaj nazwę użytkownika.
   5. W polu Hasło podaj hasło użytkownika.
   6. Kliknij pozycję Testuj połączenie w celu przetestowania połączenia.
   7. Kliknij przycisk Zapisz, aby zapisać połączoną usługę.

   

Tworzenie zestawów danych

W tym kroku utworzysz zestawy danych reprezentujące źródło danych i miejsce docelowe danych.

Tworzenie zestawu danych reprezentującego źródło danych

W tym kroku utworzysz zestaw danych reprezentujący źródło danych.

1. W widoku drzewa kliknij kolejno pozycje + (plus) i Zestaw danych.

   

2. Wybierz Azure SQL Database Managed Instance i kliknij Continue.

   

3. Na karcie Ustawianie właściwości ustaw nazwę zestawu danych i informacje o połączeniu:

   1. Wybierz AzureSqlMI1 dla Połączona usługa.
   2. Wybierz [dbo].[dbo_customers_CT] dla Nazwa tabeli. Uwaga: ta tabela została utworzona automatycznie, gdy włączono CDC w tabeli klientów. Zmienione dane nigdy nie są odpytywane bezpośrednio z tej tabeli, ale zamiast tego są wyodrębniane przy pomocy funkcji CDC.

   

Utwórz zestaw danych reprezentujący dane skopiowane do docelowego magazynu danych.

W tym kroku utworzysz zestaw danych reprezentujący dane skopiowane z magazynu danych źródłowych. Kontener data lake został utworzony w Azure Blob Storage w ramach wymagań wstępnych. Utwórz kontener, jeśli nie istnieje, lub zmień nazwę istniejącego kontenera. W tym samouczku nazwa pliku wyjściowego jest generowana dynamicznie za pomocą czasu wyzwalacza, który zostanie ustawiony później.

1. W widoku drzewa kliknij kolejno pozycje + (plus) i Zestaw danych.

   

2. Wybierz Azure Blob Storage i kliknij Continue.

   

3. Wybierz pozycję RozdzielanyTekst, a następnie kliknij przycisk Kontynuuj.

   

4. Na karcie Ustaw właściwości ustaw nazwę zestawu danych i informacje o połączeniu:

   1. Wybierz AzureStorageLinkedService dla Połączonej usługi.
   2. Wprowadź raw dla parte kontenera w filePath.
   3. Włącz pierwszy wiersz jako nagłówek
   4. Kliknij przycisk OK.

   

Tworzenie przepływu w celu skopiowania zmienionych danych

W tym kroku utworzysz potok, który najpierw sprawdza liczbę zmienionych rekordów znajdujących się w tabeli zmian przy użyciu działania lookup. Działanie warunku IF sprawdza, czy liczba zmienionych rekordów jest większa niż zero i uruchamia działanie copy skopiować wstawione/zaktualizowane/usunięte dane z Azure SQL Database do Azure Blob Storage. Na koniec skonfigurowano wyzwalacz okna przesuwnego, a czasy rozpoczęcia i zakończenia zostaną przekazane do działań jako parametry okna początkowego i końcowego.

1. W interfejsie użytkownika usługi Data Factory przejdź do karty Edycja. Kliknij + (plus) w okienku po lewej stronie, a następnie kliknij Potok.

   

2. Widzisz nową kartę do konfiguracji potoku. Widzisz też potok w widoku drzewa. W oknie Właściwości zmień nazwę potoku na IncrementalCopyPipeline.

   

3. W przyborniku Działania rozwiń pozycję Ogólne, a następnie przeciągnij działanie Lookup (Wyszukiwanie) i upuść je na powierzchni projektanta potoku. Ustaw nazwę działania na GetChangeCount. Ta czynność pobiera liczbę rekordów w tabeli zmian dla danego okna czasowego.

   

4. Przejdź do ustawień w oknie Właściwości :

   1. Określ nazwę zestawu danych SQL MI dla pola Źródłowy zestaw danych.

   2. Wybierz opcję Zapytanie i wprowadź następujące informacje w polu zapytania:

   DECLARE  @from_lsn binary(10), @to_lsn binary(10);  
   SET @from_lsn =sys.fn_cdc_get_min_lsn('dbo_customers');  
   SET @to_lsn = sys.fn_cdc_map_time_to_lsn('largest less than or equal',  GETDATE());
   SELECT count(1) changecount FROM cdc.fn_cdc_get_net_changes_dbo_customers(@from_lsn, @to_lsn, 'all')
   

   3. Włącz tylko pierwszy wiersz

   

5. Kliknij przycisk Podgląd danych, aby upewnić się, że prawidłowe dane wyjściowe są uzyskiwane przez działanie wyszukiwania

   

6. Rozwiń sekcję Iteracje i konstrukcje warunkowe w przyborniku Działania, a następnie przeciągnij i upuść akcję 'Jeśli warunek' na powierzchnię projektanta potoku. Ustaw nazwę działania na HasChangedRows.

   

7. Przejdź do działań w oknie Właściwości :

   1. Wprowadź następujące wyrażenie

   @greater(int(activity('GetChangeCount').output.firstRow.changecount),0)
   

   2. Kliknij ikonę ołówka, aby edytować warunek True.

   

   3. Rozwiń pozycję Ogólne w przyborniku Działania i dodaj działanie Wait na powierzchnię projektanta potoku. Jest to tymczasowe działanie w celu debugowania warunku If i zostanie zmienione w dalszej części samouczka.

   

   4. Kliknij ślad nawigacyjny IncrementalCopyPipeline, aby powrócić do głównego potoku.

8. Uruchom potok w trybie Debug aby sprawdzić, czy zostanie wykonany pomyślnie.

   

9. Następnie wróć do kroku True condition (Prawda) i usuń działanie Oczekiwanie. W panelu narzędzi Działania, rozwiń pozycję Przenieś & przekształć, a następnie przeciągnij i upuść operację Kopiuj na powierzchnię projektanta potoku. Ustaw nazwę działania na IncrementalCopyActivity.

   

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

11. Określ nazwę zestawu danych SQL MI dla pola Źródłowy zestaw danych.

12. Wybierz pozycję Zapytanie w polu Użyj zapytania.

13. Wprowadź następujące polecenie w polu Zapytanie.

    DECLARE @from_lsn binary(10), @to_lsn binary(10); 
    SET @from_lsn =sys.fn_cdc_get_min_lsn('dbo_customers'); 
    SET @to_lsn = sys.fn_cdc_map_time_to_lsn('largest less than or equal', GETDATE());
    SELECT * FROM cdc.fn_cdc_get_net_changes_dbo_customers(@from_lsn, @to_lsn, 'all')
    

11. Kliknij pozycję Podgląd, aby sprawdzić, czy zapytanie zwraca poprawnie zmienione wiersze.

    

12. Przejdź do karty Sink i określ zestaw danych Azure Storage dla pola Sink Dataset.

    

13. Kliknij z powrotem na główną kanwę potoku i połącz działanie Lookup z działaniem If Condition jeden po drugim. Przeciągnij zielony przycisk dołączony do działania Lookup (Wyszukiwanie) do działania If Condition (Warunek).

    

14. Na pasku narzędzi kliknij pozycję Weryfikuj. Potwierdź, że nie występują błędy walidacji. Zamknij okno Raport weryfikacji pipeline, klikając pozycję >>.

    

15. Kliknij Debuguj, aby przetestować potok i sprawdzić, czy plik jest generowany w lokalizacji przechowywania.

    

16. Opublikuj elementy (powiązane usługi, zestawy danych i potoki) w usłudze Data Factory, kliknij przycisk Opublikuj wszystko. Poczekaj na wyświetlenie komunikatu Publikowanie powiodło się.

    

Konfigurowanie wyzwalacza okna z przesuwaniem i parametrów okna CDC

W tym kroku utworzysz wyzwalacz okna z przesunięciem, aby wykonać zadanie zgodnie z częstym ustalonym harmonogramem. Użyjesz zmiennych systemowych WindowStart i WindowEnd wyzwalacza okna wirowania i przekażesz je jako parametry do potoku do użycia w zapytaniu CDC.

1. Przejdź do karty Parametry potoku IncrementalCopyPipeline i użyj przycisku + Nowy dodaj dwa parametry (triggerStartTime i triggerEndTime) do potoku, co będzie reprezentować czas rozpoczęcia i zakończenia okna wirowania. Do celów debugowania dodaj wartości domyślne w formacie RRRR-MM-DD HH24:MI:SS.FFF, ale upewnij się, że czasRozpoczęciaWyzwalacza nie jest wcześniejszy niż moment, gdy CDC zostało włączone na tabeli, w przeciwnym razie wystąpi błąd.

   

2. Kliknij kartę ustawień działania Lookup i skonfiguruj zapytanie tak, aby używało parametrów początkowych i końcowych. Skopiuj następujące elementy do zapytania:

   @concat('DECLARE @begin_time datetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); 
   SET @begin_time = ''',pipeline().parameters.triggerStartTime,''';
   SET @end_time = ''',pipeline().parameters.triggerEndTime,''';
   SET @from_lsn = sys.fn_cdc_map_time_to_lsn(''smallest greater than or equal'', @begin_time);
   SET @to_lsn = sys.fn_cdc_map_time_to_lsn(''largest less than'', @end_time);
   SELECT count(1) changecount FROM cdc.fn_cdc_get_net_changes_dbo_customers(@from_lsn, @to_lsn, ''all'')')
   

3. Przejdź do działania Kopiuj w przypadku prawdziwego działania If Condition i kliknij zakładkę Źródło. Skopiuj następujące elementy do kwerendy:

   @concat('DECLARE @begin_time datetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); 
   SET @begin_time = ''',pipeline().parameters.triggerStartTime,''';
   SET @end_time = ''',pipeline().parameters.triggerEndTime,''';
   SET @from_lsn = sys.fn_cdc_map_time_to_lsn(''smallest greater than or equal'', @begin_time);
   SET @to_lsn = sys.fn_cdc_map_time_to_lsn(''largest less than'', @end_time);
   SELECT * FROM cdc.fn_cdc_get_net_changes_dbo_customers(@from_lsn, @to_lsn, ''all'')')
   

4. Kliknij kartę Ujście działania Kopiowaniai kliknij przycisk Otwórz, aby edytować właściwości zestawu danych. Kliknij kartę Parametry i dodaj nowy parametr o nazwie triggerStart

   

5. Następnie skonfiguruj właściwości zestawu danych, aby przechowywać dane w podkatalogu klientów/przyrostowych z partycjami opartymi na datach.

   1. Kliknij kartę Połączenie właściwości zestawu danych i dodaj zawartość dynamiczną dla sekcji Katalog i Plik .

   2. Wprowadź następujące wyrażenie w sekcji Katalog, klikając łącze zawartości dynamicznej pod polem tekstowym.

      @concat('customers/incremental/',formatDateTime(dataset().triggerStart,'yyyy/MM/dd'))
      

   3. Wprowadź następujące wyrażenie w sekcji Plik . Spowoduje to utworzenie nazw plików na podstawie daty i godziny rozpoczęcia wyzwalacza, z sufiksem z rozszerzeniem csv:

      @concat(formatDateTime(dataset().triggerStart,'yyyyMMddHHmmssfff'),'.csv')
      

      

   4. Wróć do ustawień Sink w aktywności Kopiuj, klikając kartę IncrementalCopyPipeline.

   5. Rozwiń właściwości zestawu danych i wprowadź zawartość dynamiczną w wartości parametru triggerStart przy użyciu następującego wyrażenia:

      @pipeline().parameters.triggerStartTime
      

   

6. Kliknij pozycję Debuguj, aby przetestować potok i upewnić się, że struktura folderów i plik wyjściowy są generowane zgodnie z oczekiwaniami. Pobierz i otwórz plik, aby zweryfikować zawartość.

   

7. Upewnij się, że parametry są wstrzykiwane do zapytania, przeglądając parametry wejściowe przebiegu potoku.

   

8. Opublikuj elementy (powiązane usługi, zestawy danych i potoki) w usłudze Data Factory, kliknij przycisk Opublikuj wszystko. Poczekaj na wyświetlenie komunikatu Publikowanie powiodło się.

9. Na koniec skonfiguruj wyzwalacz okna przesuwnego, aby uruchamiać potok w regularnych odstępach czasowych, ustawiając parametry czasu rozpoczęcia i zakończenia.

   1. Kliknij przycisk Dodaj wyzwalacz, a następnie wybierz pozycję Nowy/Edytuj

   

   2. Wprowadź nazwę wyzwalacza i określ godzinę rozpoczęcia, która jest równa godzinie zakończenia okna debugowania powyżej.

   

   3. Na następnym ekranie określ następujące wartości odpowiednio dla parametrów początkowych i końcowych.

      @formatDateTime(trigger().outputs.windowStartTime,'yyyy-MM-dd HH:mm:ss.fff')
      @formatDateTime(trigger().outputs.windowEndTime,'yyyy-MM-dd HH:mm:ss.fff')
      

      

10. Korzystając z SQL Server Management Studio wprowadź dodatkowe zmiany w tabeli klienta, uruchamiając następujący kod SQL:

    insert into customers (customer_id, first_name, last_name, email, city) values (4, 'Farlie', 'Hadigate', 'fhadigate3@zdnet.com', 'Reading');
    insert into customers (customer_id, first_name, last_name, email, city) values (5, 'Anet', 'MacColm', 'amaccolm4@yellowbook.com', 'Portsmouth');
    insert into customers (customer_id, first_name, last_name, email, city) values (6, 'Elonore', 'Bearham', 'ebearham5@ebay.co.uk', 'Portsmouth');
    update customers set first_name='Elon' where customer_id=6;
    delete from customers where customer_id=5;
    

11. Kliknij przycisk Opublikuj wszystko. Poczekaj na wyświetlenie komunikatu Publikowanie powiodło się.

12. Po kilku minutach pipeline zostanie wyzwolony, a nowy plik zostanie załadowany do usługi Azure Storage

Monitorowanie przepływu kopiowania przyrostowego

1. Kliknij kartę Monitorowanie po lewej stronie. Na liście zostanie wyświetlone uruchomienie potoku i jego stan. Aby odświeżyć listę, kliknij pozycję Odśwież. Umieść wskaźnik myszy w pobliżu nazwy potoku danych, aby uzyskać dostęp do opcji ponownego uruchomienia i raportu zużycia.

   

2. Aby wyświetlić wykonania działań skojarzone z uruchomieniem potoku, kliknij nazwę potoku. Jeśli wykryto zmienione dane, będą istnieć trzy działania, w tym działanie kopiowania. W przeciwnym razie na liście będą znajdować się tylko dwa wpisy. Aby wrócić do widoku przebiegów potoku, kliknij link Wszystkie potoki na górze.

   

Sprawdzanie wyników

W folderze customers/incremental/YYYY/MM/DD kontenera raw widoczny będzie drugi plik.

Powiązana zawartość

Przejdź do poniższego samouczka, aby dowiedzieć się więcej o kopiowaniu nowych i zmienionych plików tylko na podstawie ich daty ostatniej modyfikacji: