Pozyskiwanie danych z usługi Azure Cosmos DB do usługi Azure Data Explorer

Usługa Azure Data Explorer obsługuje pozyskiwanie danych z usługi Azure Cosmos DB for NoSql przy użyciu zestawienia zmian. Połączenie danych zestawienia zmian usługi Cosmos DB to potok pozyskiwania, który nasłuchuje zestawienia zmian usługi Cosmos DB i pozyskuje dane do tabeli eksploratora danych. Kanał informacyjny zmian nasłuchuje nowych i zaktualizowanych dokumentów, ale nie rejestruje usuwania. Aby uzyskać ogólne informacje na temat pozyskiwania danych w usłudze Azure Data Explorer, zobacz Omówienie pozyskiwania danych w usłudze Azure Data Explorer.

Każde połączenie danych nasłuchuje określonego kontenera usługi Cosmos DB i pozyskiwa dane do określonej tabeli (więcej niż jedno połączenie może pozyskiwać w jednej tabeli). Metoda pozyskiwania obsługuje pozyskiwanie strumieniowe (po włączeniu) i pozyskiwanie w kolejce.

W tym artykule dowiesz się, jak skonfigurować połączenie danych zestawienia zmian usługi Cosmos DB w celu pozyskiwania danych do usługi Azure Data Explorer przy użyciu tożsamości zarządzanej przez system. Przed rozpoczęciem zapoznaj się z zagadnieniami .

Aby skonfigurować łącznik, wykonaj następujące czynności:

Krok 1. Wybieranie tabeli usługi Azure Data Explorer i konfigurowanie mapowania tabeli

Krok 2. Tworzenie połączenia danych usługi Cosmos DB

Krok 3. Testowanie połączenia danych

Wymagania wstępne

• Subskrypcja platformy Azure. Utwórz bezpłatne konto platformy Azure.
• Baza danych i klaster usługi Azure Data Explorer. Utwórz klaster i bazę danych.
• Kontener z konta usługi Cosmos DB dla bazy danych NoSQL.
• Jeśli konto usługi Cosmos DB blokuje dostęp sieciowy, na przykład przy użyciu prywatnego punktu końcowego, musisz utworzyć zarządzany prywatny punkt końcowy na koncie usługi Cosmos DB. Jest to wymagane, aby klaster wywoływać interfejs API zestawienia zmian.

Krok 1. Wybieranie tabeli usługi Azure Data Explorer i konfigurowanie mapowania tabeli

Przed utworzeniem połączenia danych utwórz tabelę, w której będą przechowywane pozyskane dane i zastosuje mapowanie zgodne ze schematem w źródłowym kontenerze usługi Cosmos DB. Jeśli scenariusz wymaga więcej niż prostego mapowania pól, możesz użyć zasad aktualizacji do przekształcania i mapowania danych pozyskanych z zestawienia zmian.

Poniżej przedstawiono przykładowy schemat elementu w kontenerze usługi Cosmos DB:

{
    "id": "17313a67-362b-494f-b948-e2a8e95e237e",
    "name": "Cousteau",
    "_rid": "pL0MAJ0Plo0CAAAAAAAAAA==",
    "_self": "dbs/pL0MAA==/colls/pL0MAJ0Plo0=/docs/pL0MAJ0Plo0CAAAAAAAAAA==/",
    "_etag": "\"000037fc-0000-0700-0000-626a44110000\"",
    "_attachments": "attachments/",
    "_ts": 1651131409
}

Wykonaj następujące kroki, aby utworzyć tabelę i zastosować mapowanie tabeli:

1. W internetowym interfejsie użytkownika usługi Azure Data Explorer z menu nawigacji po lewej stronie wybierz pozycję Zapytanie, a następnie wybierz bazę danych, w której chcesz utworzyć tabelę.

2. Uruchom następujące polecenie, aby utworzyć tabelę o nazwie TestTable.

   .create table TestTable(Id:string, Name:string, _ts:long, _timestamp:datetime)
   

3. Uruchom następujące polecenie, aby utworzyć mapowanie tabeli.

   Polecenie mapuje właściwości niestandardowe z dokumentu JSON usługi Cosmos DB na kolumny w tabeli TestTable w następujący sposób:

   Właściwość Cosmos DB	Kolumna tabeli	Przekształcenie
   id	Id	Brak
   name	Nazwisko	Brak
   _Ts	_ts	Brak
   _Ts	_Sygnatury czasowej	Używa DateTimeFromUnixSeconds metody do przekształcania _ts (w sekundach systemu UNIX) na _timestamp (datetime))

   Uwaga

   Zalecamy użycie następujących kolumn sygnatury czasowej:

   • _ts: użyj tej kolumny, aby uzgodnić dane z usługą Cosmos DB.
   • _timestamp: użyj tej kolumny do uruchamiania wydajnych filtrów czasu w zapytaniach Kusto. Aby uzyskać więcej informacji, zobacz Najlepsze rozwiązanie dotyczące zapytań.

   .create table TestTable ingestion json mapping "DocumentMapping"
   ```
   [
       {"column":"Id","path":"$.id"},
       {"column":"Name","path":"$.name"},
       {"column":"_ts","path":"$._ts"},
       {"column":"_timestamp","path":"$._ts", "transform":"DateTimeFromUnixSeconds"}
   ]
   ```
   

Przekształcanie i mapowania danych za pomocą zasad aktualizacji

Jeśli scenariusz wymaga więcej niż prostego mapowania pól, możesz użyć zasad aktualizacji do przekształcania i mapowania danych pozyskanych z zestawienia zmian.

Zasady aktualizacji to sposób przekształcania danych w miarę pozyskiwania ich do tabeli. Są one zapisywane w język zapytań Kusto i są uruchamiane w potoku pozyskiwania. Mogą one służyć do przekształcania danych z pozyskiwania zestawienia zmian usługi Cosmos DB, na przykład w następujących scenariuszach:

• Dokumenty zawierają tablice, które byłyby łatwiejsze do wykonywania zapytań, jeśli są przekształcane w wielu wierszach przy użyciu mv-expand operatora .
• Chcesz odfiltrować dokumenty. Na przykład można filtrować dokumenty według typu przy użyciu where operatora .
• Masz złożoną logikę, której nie można przedstawić w mapowaniu tabeli.

Aby uzyskać informacje na temat tworzenia zasad aktualizacji i zarządzania nimi, zobacz Omówienie zasad aktualizacji.

Krok 2. Tworzenie połączenia danych usługi Cosmos DB

Aby utworzyć łącznik danych, możesz użyć następujących metod:

Witryna Azure Portal

1. W witrynie Azure Portal przejdź do strony przeglądu klastra, a następnie wybierz kartę Wprowadzenie .

2. Na kafelku Pozyskiwanie danych wybierz pozycję Utwórz połączenie>danych Cosmos DB.

   

3. W okienku Tworzenie połączenia danych w usłudze Cosmos DB wypełnij formularz informacjami w tabeli:

   

   Pole	opis
   Nazwa bazy danych	Wybierz bazę danych usługi Azure Data Explorer, do której chcesz pozyskać dane.
   Nazwa połączenia danych	Określ nazwę połączenia danych.
   Subskrypcja	Wybierz subskrypcję zawierającą konto NoSQL usługi Cosmos DB.
   Konto usługi Cosmos DB	Wybierz konto usługi Cosmos DB, z którego chcesz pozyskiwać dane.
   Baza danych SQL	Wybierz bazę danych Cosmos DB, z której chcesz pozyskiwać dane.
   Kontener SQL	Wybierz kontener usługi Cosmos DB, z którego chcesz pozyskiwać dane.
   Nazwa tabeli	Określ nazwę tabeli usługi Azure Data Explorer, do której chcesz pozyskiwać dane.
   Nazwa mapowania	Opcjonalnie określ nazwę mapowania, która ma być używana dla połączenia danych.

4. Opcjonalnie w sekcji Ustawienia zaawansowane wykonaj następujące czynności:

   1. Określ datę rozpoczęcia pobierania zdarzeń. Jest to czas, od którego łącznik rozpocznie pozyskiwanie danych. Jeśli nie określisz czasu, łącznik rozpocznie pozyskiwanie danych od momentu utworzenia połączenia danych. Zalecany format daty to standard ISO 8601 UTC określony w następujący sposób: yyyy-MM-ddTHH:mm:ss.fffffffZ.

   2. Wybierz pozycję Przypisane przez użytkownika, a następnie wybierz tożsamość. Domyślnie tożsamość zarządzana przypisana przez system jest używana przez połączenie. W razie potrzeby możesz użyć tożsamości przypisanej przez użytkownika.

      

5. Wybierz pozycję Utwórz , aby utworzyć połączenie danych.

Szablon usługi ARM

Użyj następującego przykładowego szablonu usługi ARM jako podstawy do utworzenia własnego szablonu połączenia danych, a następnie wdróż go w witrynie Azure Portal.

Aby skonfigurować połączenie usługi Cosmos DB:

1. Skonfiguruj tożsamość zarządzaną systemu na potrzeby uwierzytelniania połączenia usługi Cosmos DB.

   1. W internetowym interfejsie użytkownika usługi Azure Data Explorer wybierz pozycję Zapytanie z menu nawigacji po lewej stronie, a następnie wybierz klaster lub bazę danych dla połączenia danych.

2. Udziel uprawnienia połączenia danych w celu uzyskania dostępu do konta usługi Cosmos DB. Zapewnienie dostępu do połączenia danych z usługą Cosmos DB umożliwia mu uzyskiwanie dostępu do danych i pobieranie ich z bazy danych. Potrzebny będzie identyfikator podmiotu zabezpieczeń klastra, który można znaleźć w witrynie Azure Portal. Aby uzyskać więcej informacji, zobacz Konfigurowanie tożsamości zarządzanych dla klastra.

   Uwaga

   • Następujące kroki umożliwiają przypisanie tych ról do identyfikatora podmiotu zabezpieczeń:
     • Wbudowany czytnik danych usługi Cosmos DB
       • Nie można przypisać roli Wbudowanego czytnika danych usługi Cosmos DB przy użyciu funkcji przypisywania ról w witrynie Azure Portal.
     • Rola czytelnika konta usługi Cosmos DB

   Użyj jednej z następujących opcji, aby udzielić dostępu do konta usługi Cosmos DB:

   • Udziel dostępu przy użyciu interfejsu wiersza polecenia platformy Azure: uruchom polecenie interfejsu wiersza polecenia, korzystając z informacji w poniższej tabeli, aby zastąpić symbole zastępcze odpowiednimi wartościami:

     az cosmosdb sql role assignment create --account-name <CosmosDbAccountName> --resource-group <CosmosDbResourceGroup> --role-definition-id 00000000-0000-0000-0000-000000000001 --principal-id <ClusterPrincipalId> --scope "/"
     
     az role assignment create --role fbdf93bf-df7d-467e-a4d2-9458aa1360c8 --assignee <ClusterPrincipalId> --scope <CosmosDBAccountResourceId>
     

     Symbol zastępczy	opis
     <CosmosDBAccountName>	Nazwa konta usługi Cosmos DB.
     <CosmosDBResourceGroup>	Nazwa grupy zasobów zawierającej konto usługi Cosmos DB.
     <CosmosDBAccountResourceId>	Identyfikator zasobu platformy Azure (począwszy od subscriptions/) konta usługi Cosmos DB.
     <ClusterPrincipalId>	Identyfikator podmiotu zabezpieczeń tożsamości zarządzanej przypisanej do klastra. Identyfikator reguły klastra można znaleźć w witrynie Azure Portal. Aby uzyskać więcej informacji, zobacz Konfigurowanie tożsamości zarządzanych dla klastra.

   • Udziel dostępu przy użyciu szablonu usługi ARM: wdróż następujący szablon w grupie zasobów konta usługi Cosmos DB:

     {
         "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
         "contentVersion": "1.0.0.0",
         "parameters": {
             "clusterPrincipalId": {
                 "type": "string",
                 "metadata": { "description": "The principle ID of your cluster." }
             },
             "cosmosDbAccount": {
                 "type": "string",
                 "metadata": { "description": "The name of your Cosmos DB account." }
             },
             "cosmosDbAccountResourceId": {
                 "type": "string",
                 "metadata": { "description": "The resource ID of your Cosmos DB account." }
             }
         },
         "variables": {
             "cosmosDataReader": "00000000-0000-0000-0000-000000000001",
             "dataRoleDefinitionId": "[format('/subscriptions/{0}/resourceGroups/{1}/providers/Microsoft.DocumentDB/databaseAccounts/{2}/sqlRoleDefinitions/{3}', subscription().subscriptionId, resourceGroup().name, parameters('cosmosDbAccount'), variables('cosmosDataReader'))]",
             "roleAssignmentId": "[guid(parameters('cosmosDbAccountResourceId'), parameters('clusterPrincipalId'))]",
             "rbacRoleDefinitionId": "[format('/subscriptions/{0}/providers/Microsoft.Authorization/roleDefinitions/{1}', subscription().subscriptionId, 'fbdf93bf-df7d-467e-a4d2-9458aa1360c8')]"
         },
         "resources": [
             {
                 "type": "Microsoft.DocumentDB/databaseAccounts/sqlRoleAssignments",
                 "apiVersion": "2022-08-15",
                 "name": "[concat(parameters('cosmosDbAccount'), '/', guid(parameters('clusterPrincipalId'), parameters('cosmosDbAccount')))]",
                 "properties": {
                     "principalId": "[parameters('clusterPrincipalId')]",
                     "roleDefinitionId": "[variables('dataRoleDefinitionId')]",
                     "scope": "[resourceId('Microsoft.DocumentDB/databaseAccounts', parameters('cosmosDbAccount'))]"
                 }
             },
             {
                 "type": "Microsoft.Authorization/roleAssignments",
                 "apiVersion": "2022-04-01",
                 "name": "[variables('roleAssignmentId')]",
                 "scope": "[format('Microsoft.DocumentDb/databaseAccounts/{0}', parameters('cosmosDbAccount'))]",
                 "properties": {
                     "description": "Giving RBAC reader on Cosmos DB",
                     "principalId": "[parameters('clusterPrincipalId')]",
                     "principalType": "ServicePrincipal",
                     "roleDefinitionId": "[variables('rbacRoleDefinitionId')]"
                 }
             }
         ]
     }
     

3. Wdróż następujący szablon usługi ARM, aby utworzyć połączenie danych usługi Cosmos DB. Zastąp symbole zastępcze odpowiednimi wartościami.

   {
     "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
     "contentVersion": "1.0.0.0",
     "parameters": {
       "kustoClusterName": {
         "type": "string",
         "metadata": { "description": "Kusto Cluster name" }
       },
       "kustoDbName": {
         "type": "string",
         "metadata": { "description": "Kusto Database name" }
       },
       "kustoConnectionName": {
         "type": "string",
         "metadata": { "description": "Kusto Database connection name" }
       },
       "kustoLocation": {
         "type": "string",
         "metadata": { "description": "Location (Azure Region) of the Kusto cluster" }
       },
       "kustoTable": {
         "type": "string",
         "metadata": { "description": "Kusto Table name where to ingest data" }
       },
       "kustoMappingRuleName": {
         "type": "string",
         "defaultValue": "",
         "metadata": { "description": "Mapping name of the Kusto Table (if omitted, default mapping is applied)" }
       },
       "managedIdentityResourceId": {
         "type": "string",
         "metadata": { "description": "ARM resource ID of the managed identity (cluster resource ID for system or user identity)" }
       },
       "cosmosDbAccountResourceId": {
         "type": "string",
         "metadata": { "description": "ARM resource ID of Cosoms DB account" }
       },
       "cosmosDbDatabase": {
         "type": "string",
         "metadata": { "description": "Cosmos DB Database name" }
       },
       "cosmosDbContainer": {
         "type": "string",
         "metadata": { "description": "Cosmos DB container name" }
       },
       "retrievalStartDate": {
         "type": "string",
         "defaultValue": "",
         "metadata": { "description": "Date-time at which to start the data retrieval; default: 'now' if not provided. Recommended format: yyyy-MM-ddTHH:mm:ss.fffffffZ" }
       }
     },
     "variables": { },
     "resources": [{
       "type": "Microsoft.Kusto/Clusters/Databases/DataConnections",
       "apiVersion": "2022-11-11",
       "name": "[concat(parameters('kustoClusterName'), '/', parameters('kustoDbName'), '/', parameters('kustoConnectionName'))]",
       "location": "[parameters('kustoLocation')]",
       "kind": "CosmosDb",
       "properties": {
         "tableName": "[parameters('kustoTable')]",
         "mappingRuleName": "[parameters('kustoMappingRuleName')]",
         "managedIdentityResourceId": "[parameters('managedIdentityResourceId')]",
         "cosmosDbAccountResourceId": "[parameters('cosmosDbAccountResourceId')]",
         "cosmosDbDatabase": "[parameters('cosmosDbDatabase')]",
         "cosmosDbContainer": "[parameters('cosmosDbContainer')]",
         "retrievalStartDate": "[parameters('retrievalStartDate')]"
       }
     }]
   }
   

Krok 3. Testowanie połączenia danych

1. W kontenerze usługi Cosmos DB wstaw następujący dokument:

   {
       "name":"Cousteau"
   }
   

2. W internetowym interfejsie użytkownika usługi Azure Data Explorer uruchom następujące zapytanie:

   TestTable
   

   Zestaw wyników powinien wyglądać podobnie do poniższej ilustracji:

   

Uwaga

Usługa Azure Data Explorer ma zasady agregacji (dzielenia na partie) na potrzeby pozyskiwania danych w kolejce przeznaczone do optymalizacji procesu pozyskiwania danych. Domyślne zasady dzielenia na partie są skonfigurowane do uszczelniania partii, gdy jeden z następujących warunków jest spełniony w partii: maksymalny czas opóźnienia wynoszący 5 minut, całkowity rozmiar jednego GB lub 1000 obiektów blob. W związku z tym może wystąpić opóźnienie. Aby uzyskać więcej informacji, zobacz zasady dzielenia na partie. Aby zmniejszyć opóźnienie, skonfiguruj tabelę do obsługi przesyłania strumieniowego. Zobacz zasady przesyłania strumieniowego.

Kwestie wymagające rozważenia

Następujące zagadnienia dotyczą zestawienia zmian usługi Cosmos DB:

• Kanał informacyjny zmian nie ujawnia zdarzeń usuwania .

  Zestawienie zmian usługi Cosmos DB zawiera tylko nowe i zaktualizowane dokumenty. Jeśli musisz wiedzieć o usuniętych dokumentach, możesz skonfigurować zestawienie przy użyciu znacznika nietrwałego , aby oznaczyć dokument usługi Cosmos DB jako usunięty. Właściwość jest dodawana do aktualizacji zdarzeń wskazujących, czy dokument został usunięty. Następnie możesz użyć where operatora w zapytaniach, aby je odfiltrować.

  Jeśli na przykład zamapujesz usuniętą właściwość na kolumnę tabeli o nazwie IsDeleted, możesz odfiltrować usunięte dokumenty przy użyciu następującego zapytania:

  TestTable
  | where not(IsDeleted)
  

• Kanał informacyjny zmian uwidacznia tylko najnowszą aktualizację dokumentu.

  Aby zrozumieć konsekwencje drugiego zagadnienia, zapoznaj się z następującym scenariuszem:

  Kontener usługi Cosmos DB zawiera dokumenty A i B. Zmiany właściwości o nazwie foo są wyświetlane w poniższej tabeli:

  Identyfikator dokumentu	Obiekt foo	Zdarzenie	Sygnatura czasowa dokumentu (_ts)
  A	Czerwony	Tworzenie	10
  B	Blue (Niebieski)	Tworzenie	20
  A	Orange	Zaktualizuj	30
  A	Różowy	Update	40
  B	Fiołek	Zaktualizuj	50
  A	Karminowy	Zaktualizuj	50
  B	NeonBlue	Zaktualizuj	70

  Interfejs API zestawienia zmian jest sondowany przez łącznik danych w regularnych odstępach czasu, zazwyczaj co kilka sekund. Każda ankieta zawiera zmiany, które wystąpiły w kontenerze między wywołaniami, ale tylko najnowszą wersją zmiany na dokument.

  Aby zilustrować problem, rozważ sekwencję wywołań interfejsu API ze znacznikami czasu 15, 35, 55 i 75, jak pokazano w poniższej tabeli:

  Sygnatura czasowa wywołania interfejsu API	Identyfikator dokumentu	Obiekt foo	Sygnatura czasowa dokumentu (_ts)
  15	A	Czerwony	10
  35	B	Blue (Niebieski)	20
  35	A	Orange	30
  55	B	Fiołek	50
  55	A	Karminowy	60
  75	B	NeonBlue	70

  Porównanie wyników interfejsu API z listą zmian wprowadzonych w dokumencie usługi Cosmos DB zauważysz, że nie są one zgodne. Zdarzenie aktualizacji dokumentu A, wyróżnione w tabeli zmian w znaczniku czasu 40, nie jest wyświetlane w wynikach wywołania interfejsu API.

  Aby zrozumieć, dlaczego zdarzenie nie jest wyświetlane, przeanalizujemy zmiany dokumentu A między wywołaniami interfejsu API w znacznikach czasu 35 i 55. Między tymi dwoma wywołaniami dokument A uległ zmianie dwa razy w następujący sposób:

  Identyfikator dokumentu	Obiekt foo	Zdarzenie	Sygnatura czasowa dokumentu (_ts)
  A	Różowy	Zaktualizuj	40
  A	Karminowy	Zaktualizuj	50

  Po wprowadzeniu wywołania interfejsu API znacznika czasu 55 interfejs API zestawienia zmian zwraca najnowszą wersję dokumentu. W tym przypadku najnowsza wersja dokumentu A to aktualizacja sygnatury czasowej 50, która jest aktualizacją foo właściwości od Pink do Carmine.

  W związku z tym scenariuszem łącznik danych może przegapić pewne pośrednie zmiany dokumentu. Na przykład niektóre zdarzenia mogą zostać pominięte, jeśli usługa połączenia danych nie działa przez kilka minut lub częstotliwość zmian dokumentu jest wyższa niż częstotliwość sondowania interfejsu API. Przechwytywany jest jednak najnowszy stan każdego dokumentu.

• Usuwanie i ponowne tworzenie kontenera usługi Cosmos DB nie jest obsługiwane

  Usługa Azure Data Explorer śledzi zestawienie zmian, wskazując „pozycję”, w której znajduje się zestawienie. Odbywa się to przy użyciu tokenu kontynuacji na każdej fizycznej partycji kontenera. Po usunięciu/ponownym utworzeniu kontenera te tokeny kontynuacji są nieprawidłowe i nie są resetowane: należy usunąć i ponownie utworzyć połączenie danych.

Szacowanie kosztów

Ile korzystasz z połączenia danych usługi Cosmos DB, wpływa na użycie jednostek żądań (RU) kontenera usługi Cosmos DB?

Łącznik wywołuje interfejs API zestawienia zmian usługi Cosmos DB na każdej partycji fizycznej kontenera do maksymalnie raz na sekundę. Następujące koszty są skojarzone z tymi wywołaniami:

Koszt	opis
Koszty stałe	Koszty stałe to około 2 jednostek RU na partycję fizyczną co sekundę.
Zmienne koszty	Koszty zmienne to około 2% jednostek RU używanych do zapisywania dokumentów, ale może się to różnić w zależności od scenariusza. Jeśli na przykład zapisujesz 100 dokumentów w kontenerze usługi Cosmos DB, koszt pisania tych dokumentów wynosi 1000 jednostek RU. Odpowiedni koszt użycia łącznika do odczytania tego dokumentu wynosi około 2% kosztów zapisu, czyli około 20 jednostek RU.

Powiązana zawartość

• Pobierz najnowsze wersje dokumentów usługi Azure Cosmos DB
• Omówienie języka Kusto Query Language (KQL)