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 pozysuje dane do tabeli Data Explorer. 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 pozyskuje dane w 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 systemu. Przed rozpoczęciem zapoznaj się z zagadnieniami .

Aby skonfigurować łącznik, wykonaj następujące kroki:

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

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 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 zastosuj 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łcenia
   id	Id	Brak
   Nazwa	Nazwa	Brak
   _Ts	_ts	Brak
   _Ts	_Sygnatury czasowej	Używa DateTimeFromUnixSeconds do przekształcania_ts (w sekundach systemu UNIX) do _timestamp (datetime))

   Uwaga

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

   • _ts: ta kolumna służy do uzgadniania danych z usługą Cosmos DB.
   • _timestamp: użyj tej kolumny do uruchamiania filtrów czasu efektywnego 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ą 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 wiele wierszy 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żna użyć następujących metod:

Witryna Azure Portal

1. W 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> danychCosmos 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 pozyskać dane.
   Baza danych SQL	Wybierz bazę danych Cosmos DB, z której chcesz pozyskać dane.
   Kontener SQL	Wybierz kontener usługi Cosmos DB, z którego chcesz pozyskać 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 tworzenia własnego szablonu połączenia danych, a następnie wdróż go w 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. Przyznaj połączeniu danych uprawnienie dostępu do konta usługi Cosmos DB. Zapewnienie dostępu do połączenia danych z usługą Cosmos DB umożliwia mu dostęp do danych i pobieranie ich z bazy danych. Potrzebny będzie identyfikator podmiotu zabezpieczeń klastra, który można znaleźć w 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 Wbudowany czytelnik danych usługi Cosmos DB przy użyciu funkcji przypisywania ról 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 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ć jak na 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. Domyślne zasady dzielenia na partie są skonfigurowane do uszczelniania partii, gdy jeden z następujących warunków jest spełniony dla 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.

Zagadnienia do 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 zdarzeń aktualizacji, które wskazują, 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 przedstawiono w poniższej tabeli:

  Identyfikator dokumentu	Obiekt foo	Zdarzenie	Sygnatura czasowa dokumentu (_ts)
  A	Red	Tworzenie	10
  B	Blue (Niebieski)	Tworzenie	20
  A	Orange	Aktualizacja	30
  A	Różowy	Aktualizowanie	40
  B	Fioletowy	Aktualizacja	50
  A	Carmine	Aktualizacja	50
  B	NeonBlue	Aktualizacja	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	Red	10
  35	B	Blue (Niebieski)	20
  35	A	Orange	30
  55	B	Fioletowy	50
  55	A	Carmine	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 Dwa razy zmienił się w następujący sposób:

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

  Po wprowadzeniu wywołania interfejsu API sygnatury czasowej 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ą właściwości foo z Pink do Carmine.

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

• 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 korzysta z połączenia danych usługi Cosmos DB, ma wpływ 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:

Koszty	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 pisania dokumentów, ale może się to różnić w zależności od scenariusza. Jeśli na przykład napiszesz 100 dokumentów do kontenera 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 ich zapisu, około 20 jednostek RU.

Zawartość pokrewna

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