Udostępnij za pośrednictwem

Wprowadzenie do przechwytywania zmian danych w magazynie analitycznym dla usługi Azure Cosmos DB

  • Artykuł

DOTYCZY: NoSQL MongoDB

Użyj funkcji przechwytywania zmian danych (CDC) w magazynie analitycznym usługi Azure Cosmos DB jako źródła do usługi Azure Data Factory lub Azure Synapse Analytics , aby przechwycić określone zmiany danych.

Uwaga

Należy pamiętać, że połączony interfejs usługi dla interfejsu API usługi Azure Cosmos DB dla bazy danych MongoDB nie jest jeszcze dostępny w przepływie danych. Jednak możesz użyć punktu końcowego dokumentu konta z połączonym interfejsem usługi "Azure Cosmos DB for NoSQL" jako obejściem, dopóki połączona usługa Mongo nie będzie obsługiwana. W połączonej usłudze NoSQL wybierz pozycję "Wprowadź ręcznie", aby podać informacje o koncie usługi Cosmos DB i użyć punktu końcowego dokumentu konta (np. https://[your-database-account-uri].documents.azure.com:443/) zamiast punktu końcowego bazy danych MongoDB (np. mongodb://[your-database-account-uri].mongo.cosmos.azure.com:10255/)

Wymagania wstępne

Włączanie magazynu analitycznego

Najpierw włącz usługę Azure Synapse Link na poziomie konta, a następnie włącz magazyn analityczny dla kontenerów odpowiednich dla obciążenia.

  1. Włączanie usługi Azure Synapse Link: włączanie usługi Azure Synapse Link dla konta usługi Azure Cosmos DB

  2. Włącz magazyn analityczny dla kontenerów:

    Opcja Przewodnik
    Włączanie dla określonego nowego kontenera Włączanie usługi Azure Synapse Link dla nowych kontenerów
    Włączanie dla określonego istniejącego kontenera Włączanie usługi Azure Synapse Link dla istniejących kontenerów

Tworzenie docelowego zasobu platformy Azure przy użyciu przepływów danych

Funkcja przechwytywania danych zmian magazynu analitycznego jest dostępna za pośrednictwem funkcji przepływu danych usługi Azure Data Factory lub Azure Synapse Analytics. W tym przewodniku użyj usługi Azure Data Factory.

Ważne

Alternatywnie możesz użyć usługi Azure Synapse Analytics. Najpierw utwórz obszar roboczy usługi Azure Synapse, jeśli jeszcze go nie masz. W nowo utworzonym obszarze roboczym wybierz kartę Programowanie , wybierz pozycję Dodaj nowy zasób, a następnie wybierz pozycję Przepływ danych.

  1. Utwórz usługę Azure Data Factory, jeśli jeszcze jej nie masz.

    Napiwek

    Jeśli to możliwe, utwórz fabrykę danych w tym samym regionie, w którym znajduje się twoje konto usługi Azure Cosmos DB.

  2. Uruchom nowo utworzoną fabrykę danych.

  3. W fabryce danych wybierz kartę Przepływy danych, a następnie wybierz pozycję Nowy przepływ danych.

  4. Nadaj nowo utworzonemu przepływowi danych unikatową nazwę. W tym przykładzie przepływ danych nosi nazwę cosmoscdc.

    Zrzut ekranu przedstawiający nowy przepływ danych o nazwie cosmoscdc.

Konfigurowanie ustawień źródła dla kontenera magazynu analitycznego

Teraz utwórz i skonfiguruj źródło do przepływu danych z magazynu analitycznego konta usługi Azure Cosmos DB.

  1. Wybierz Dodaj źródło danych.

    Zrzut ekranu przedstawiający opcję menu Dodaj źródło.

  2. W polu Nazwa strumienia wyjściowego wprowadź cosmos.

    Zrzut ekranu przedstawiający nazewnictwo nowo utworzonego źródłowego cosmosu.

  3. W sekcji Typ źródła wybierz pozycję Wbudowany.

    Zrzut ekranu przedstawiający wybieranie typu źródła wbudowanego.

  4. W polu Zestaw danych wybierz pozycję Azure — Azure Cosmos DB for NoSQL.

    Zrzut ekranu przedstawiający wybieranie usługi Azure Cosmos DB for NoSQL jako typu zestawu danych.

  5. Utwórz nową połączoną usługę dla swojego konta o nazwie cosmoslinkedservice. Wybierz istniejące konto usługi Azure Cosmos DB for NoSQL w oknie podręcznym Nowa połączona usługa , a następnie wybierz przycisk OK. W tym przykładzie wybieramy istniejące konto usługi Azure Cosmos DB for NoSQL o nazwie i bazę danych o nazwie msdocs-cosmos-source cosmicworks.

    Zrzut ekranu przedstawiający okno dialogowe Nowa połączona usługa z wybranym kontem usługi Azure Cosmos DB.

  6. Wybierz pozycję Analiza dla typu magazynu.

    Zrzut ekranu przedstawiający opcję analityczną wybraną dla połączonej usługi.

  7. Wybierz kartę Opcje źródła.

  8. W obszarze Opcje źródła wybierz kontener docelowy i włącz debugowanie przepływu danych. W tym przykładzie kontener ma nazwę products.

    Zrzut ekranu przedstawiający kontener źródłowy wybrany nazwanych produktów.

  9. Wybierz pozycję Debugowanie przepływu danych. W oknie podręcznym Włączanie debugowania przepływu danych zachowaj opcje domyślne, a następnie wybierz przycisk OK.

    Zrzut ekranu przedstawiający opcję przełączania w celu włączenia debugowania przepływu danych.

  10. Karta Opcje źródła zawiera również inne opcje, które można włączyć. W tej tabeli opisano te opcje:

Opcja Opis
Przechwytywanie aktualizacji pośrednich Włącz tę opcję, jeśli chcesz przechwycić historię zmian w elementach, w tym pośrednie zmiany między odczytami przechwytywania zmian danych.
Przechwytywanie usuniętych Włącz tę opcję, aby przechwycić rekordy usunięte przez użytkownika i zastosować je w ujściu. Nie można zastosować usuwania w eksploratorze danych platformy Azure i ujściach usługi Azure Cosmos DB.
Przechwytywanie list TTL magazynu transakcyjnego Włącz tę opcję, aby przechwycić rekordy czasu wygaśnięcia (czas wygaśnięcia) magazynu transakcyjnego usługi Azure Cosmos DB i zastosować je w ujściu. Nie można zastosować usuwania czasu wygaśnięcia w ujściach usługi Azure Data Explorer i azure Cosmos DB.
Rozmiar wsadowy w bajtach To ustawienie jest w rzeczywistości gigabajtami. Określ rozmiar w gigabajtach, jeśli chcesz wsadować zestawienia przechwytywania zmian danych
Dodatkowe konfiguracje Dodatkowe konfiguracje magazynu analitycznego usługi Azure Cosmos DB i ich wartości. (np. spark.cosmos.allowWhiteSpaceInFieldNames -> true)

Praca z opcjami źródła

Po sprawdzeniu dowolnej z opcji , Capture Deltesi Capture Transactional store TTLs proces CDC utworzy i wypełni __usr_opType pole w ujściu Capture intermediate updatesnastępującymi wartościami:

Wartość Opis Opcja
1 UPDATE Przechwyć aktualizacje pośrednie
2 INSERT Nie ma opcji wstawiania, jest domyślnie włączona
3 USER_DELETE Przechwytywanie usuniętych
100 TTL_DELETE Przechwytywanie list TTL magazynu transakcyjnego

Jeśli musisz odróżnić rekordy czasu wygaśnięcia usunięte z dokumentów usuniętych przez użytkowników lub aplikacje, sprawdź obie Capture intermediate updates opcje i Capture Transactional store TTLs . Następnie musisz dostosować procesy cdC lub aplikacje lub zapytania do użycia __usr_opType zgodnie z potrzebami biznesowymi.

Napiwek

Jeśli istnieje potrzeba, aby odbiorcy podrzędni przywrócili kolejność aktualizacji z zaznaczoną opcją "przechwyć aktualizacje pośrednie", pole sygnatury _ts czasowej systemu może być używane jako pole zamawiania.

Tworzenie i konfigurowanie ustawień ujścia dla operacji aktualizacji i usuwania

Najpierw utwórz prosty ujście usługi Azure Blob Storage , a następnie skonfiguruj ujście, aby filtrować dane tylko do określonych operacji.

  1. Utwórz konto i kontener usługi Azure Blob Storage , jeśli jeszcze go nie masz. W następnych przykładach użyjemy konta o nazwie msdocsblobstorage i kontenera o nazwie output.

    Napiwek

    Jeśli to możliwe, utwórz konto magazynu w tym samym regionie, w którym znajduje się twoje konto usługi Azure Cosmos DB.

  2. Wróć do usługi Azure Data Factory, utwórz nowy ujście dla danych zmian przechwyconych ze cosmos źródła.

    Zrzut ekranu przedstawiający dodawanie nowego ujścia połączonego z istniejącym źródłem.

  3. Nadaj ujściu unikatową nazwę. W tym przykładzie ujście nosi nazwę storage.

    Zrzut ekranu przedstawiający nazewnictwo nowo utworzonego magazynu ujścia.

  4. W sekcji Typ ujścia wybierz pozycję Wbudowany. W polu Zestaw danych wybierz pozycję Delta.

    Zrzut ekranu przedstawiający wybieranie typu zestawu danych delta wbudowanego dla ujścia.

  5. Utwórz nową połączoną usługę dla konta przy użyciu usługi Azure Blob Storage o nazwie storagelinkedservice. Wybierz istniejące konto usługi Azure Blob Storage w oknie podręcznym Nowa połączona usługa , a następnie wybierz przycisk OK. W tym przykładzie wybieramy wstępnie istniejące konto usługi Azure Blob Storage o nazwie msdocsblobstorage.

    Zrzut ekranu przedstawiający opcje typu usługi dla nowej połączonej usługi delta.

    Zrzut ekranu przedstawiający okno dialogowe Nowa połączona usługa z wybranym kontem usługi Azure Blob Storage.

  6. Wybierz kartę Ustawienia.

  7. W obszarze Ustawienia ustaw ścieżkę folderu na nazwę kontenera obiektów blob. W tym przykładzie nazwa kontenera to output.

    Zrzut ekranu przedstawiający kontener obiektów blob o nazwie output set jako obiekt docelowy ujścia.

  8. Znajdź sekcję Metoda aktualizacji i zmień wybrane opcje, aby zezwalać tylko na operacje usuwania i aktualizacji . Ponadto określ kolumny Klucz jako listę kolumn używających pola {_rid} jako unikatowego identyfikatora.

    Zrzut ekranu przedstawiający metody aktualizacji i kolumny kluczy określone dla ujścia.

  9. Wybierz pozycję Zweryfikuj , aby upewnić się, że nie wystąpiły żadne błędy ani pominięte. Następnie wybierz pozycję Publikuj , aby opublikować przepływ danych.

    Zrzut ekranu przedstawiający opcję weryfikowania, a następnie publikowania bieżącego przepływu danych.

Planowanie wykonywania przechwytywania zmian danych

Po opublikowaniu przepływu danych możesz dodać nowy potok, aby przenieść i przekształcić dane.

  1. Tworzenie nowego potoku. Nadaj potokowi unikatową nazwę. W tym przykładzie potok ma nazwę cosmoscdcpipeline.

    Zrzut ekranu przedstawiający nową opcję potoku w sekcji zasobów.

  2. W sekcji Działania rozwiń opcję Przenieś i przekształć, a następnie wybierz pozycję Przepływ danych.

    Zrzut ekranu przedstawiający opcję działania przepływu danych w sekcji działań.

  3. Nadaj działaniu przepływu danych unikatową nazwę. W tym przykładzie działanie nosi nazwę cosmoscdcactivity.

  4. Na karcie Ustawienia wybierz przepływ danych o nazwie cosmoscdc utworzony wcześniej w tym przewodniku. Następnie wybierz rozmiar obliczeniowy na podstawie ilości danych i wymaganego opóźnienia dla obciążenia.

    Zrzut ekranu przedstawiający ustawienia konfiguracji zarówno dla przepływu danych, jak i rozmiaru obliczeniowego działania.

    Napiwek

    W przypadku przyrostowych rozmiarów danych większych niż 100 GB zalecamy rozmiar niestandardowy z liczbą rdzeni 32 (+16 rdzeni sterowników).

  5. Wybierz pozycję Dodaj wyzwalacz. Zaplanuj wykonanie tego potoku w tempie, który ma sens dla obciążenia. W tym przykładzie potok jest skonfigurowany do wykonywania co pięć minut.

    Zrzut ekranu przedstawiający przycisk dodawania wyzwalacza dla nowego potoku.

    Zrzut ekranu przedstawiający konfigurację wyzwalacza na podstawie harmonogramu, począwszy od roku 2023, który jest uruchamiany co pięć minut.

    Uwaga

    Minimalne okno cyklu dla wykonań przechwytywania danych zmian wynosi minutę.

  6. Wybierz pozycję Zweryfikuj , aby upewnić się, że nie wystąpiły żadne błędy ani pominięte. Następnie wybierz pozycję Publikuj , aby opublikować potok.

  7. Obserwuj dane umieszczone w kontenerze usługi Azure Blob Storage jako dane wyjściowe przepływu danych przy użyciu funkcji przechwytywania zmian magazynu analitycznego usługi Azure Cosmos DB.

    Zrzut ekranu przedstawiający pliki wyjściowe z potoku w kontenerze usługi Azure Blob Storage.

    Uwaga

    Początkowy czas uruchamiania klastra może potrwać do trzech minut. Aby uniknąć czasu uruchamiania klastra w kolejnych wykonaniach przechwytywania danych zmiany, skonfiguruj wartość Czas wygaśnięcia klastra przepływu danych. Aby uzyskać więcej informacji na temat środowiska uruchomieniowego itegration i czasu wygaśnięcia, zobacz Integration Runtime w usłudze Azure Data Factory.

Współbieżne zadania

Rozmiar partii w opcjach źródłowych lub sytuacjach, gdy ujście jest powolne pozyskiwanie strumienia zmian, może spowodować wykonanie wielu zadań w tym samym czasie. Aby uniknąć tej sytuacji, ustaw opcję współbieżności na 1 w ustawieniach potoku, aby upewnić się, że nowe wykonania nie zostaną wyzwolone do momentu zakończenia bieżącego wykonywania.

Następne kroki