Mata in data från Azure Cosmos DB till Azure Data Explorer

Azure Data Explorer stöder datainmatning från Azure Cosmos DB for NoSql med hjälp av en ändringsfeed. Cosmos DB-dataanslutningen för ändringsflöde är en inmatningspipeline som lyssnar på cosmos DB-ändringsflödet och matar in data i din Data Explorer-tabell. Ändringsflödet lyssnar efter nya och uppdaterade dokument men loggar inte borttagna. Allmän information om datainmatning i Azure Data Explorer finns i Översikt över Datainmatning i Azure Data Explorer.

Varje dataanslutning lyssnar på en specifik Cosmos DB-container och matar in data i en angiven tabell (mer än en anslutning kan mata in i en enda tabell). Inmatningsmetoden stöder strömningsinmatning (när den är aktiverad) och köad inmatning.

I den här artikeln får du lära dig hur du konfigurerar en Cosmos DB-dataanslutning för ändringsflöde för att mata in data i Azure Data Explorer med systemhanterad identitet. Granska övervägandena innan du börjar.

Använd följande steg för att konfigurera en anslutningsapp:

Steg 1: Välj en Azure Data Explorer-tabell och konfigurera dess tabellmappning

Steg 2: Skapa en Cosmos DB-dataanslutning

Steg 3: Testa dataanslutningen

Förutsättningar

• En Azure-prenumeration. Skapa ett kostnadsfritt Azure-konto.
• Ett Azure Data Explorer-kluster och en databas. Skapa ett kluster och en databas.
• En container från ett Cosmos DB-konto för NoSQL.
• Om ditt Cosmos DB-konto blockerar nätverksåtkomst, till exempel med hjälp av en privat slutpunkt, måste du skapa en hanterad privat slutpunkt till Cosmos DB-kontot. Detta krävs för att klustret ska kunna anropa API:et för ändringsflöde.

Steg 1: Välj en Azure Data Explorer-tabell och konfigurera dess tabellmappning

Innan du skapar en dataanslutning skapar du en tabell där du lagrar inmatade data och tillämpar en mappning som matchar schemat i Cosmos DB-källcontainern. Om ditt scenario kräver mer än en enkel mappning av fält kan du använda uppdateringsprinciper för att transformera och mappa data som matas in från ändringsflödet.

Följande visar ett exempelschema för ett objekt i Cosmos DB-containern:

{
    "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
}

Använd följande steg för att skapa en tabell och använda en tabellmappning:

1. Välj Fråga på den vänstra navigeringsmenyn i Azure Data Explorer webbgränssnitt och välj sedan den databas där du vill skapa tabellen.

2. Kör följande kommando för att skapa en tabell med namnet TestTable.

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

3. Kör följande kommando för att skapa tabellmappningen.

   Kommandot mappar anpassade egenskaper från ett Cosmos DB JSON-dokument till kolumner i TestTable-tabellen enligt följande:

   Cosmos DB-egenskap	Tabellkolumn	Datatransformering
   id	Id	Ingen
   Namn	Name	Ingen
   _Ts	_ts	Ingen
   _Ts	_Tidsstämpel	Använder DateTimeFromUnixSeconds för att transformera_ts (UNIX-sekunder) till _timestamp (datetime))

   Anteckning

   Vi rekommenderar att du använder följande tidsstämpelkolumner:

   • _ts: Använd den här kolumnen för att stämma av data med Cosmos DB.
   • _timestamp: Använd den här kolumnen för att köra effektiva tidsfilter i Kusto-frågor. Mer information finns i Bästa praxis för frågor.

   .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"}
   ]
   ```
   

Transformera och mappa data med uppdateringsprinciper

Om ditt scenario kräver mer än en enkel mappning av fält kan du använda uppdateringsprinciper för att transformera och mappa data som matas in från ändringsflödet.

Uppdateringsprinciper är ett sätt att transformera data när de matas in i tabellen. De skrivs i Kusto-frågespråk och körs på inmatningspipelinen. De kan användas för att transformera data från en Cosmos DB-ändringsflödesinmatning, till exempel i följande scenarier:

• Dokumenten innehåller matriser som skulle vara enklare att fråga om de transformeras på flera rader med operatorn mv-expand .
• Du vill filtrera bort dokument. Du kan till exempel filtrera bort dokument efter typ med operatorn where .
• Du har komplex logik som inte kan representeras i en tabellmappning.

Information om hur du skapar och hanterar uppdateringsprinciper finns i Översikt över uppdateringsprinciper.

Steg 2: Skapa en Cosmos DB-dataanslutning

Du kan använda följande metoder för att skapa dataanslutningsappen:

Azure-portalen

1. I Azure Portal går du till översiktssidan för klustret och väljer sedan fliken Komma igång.

2. På panelen Datainmatning väljer du Skapa dataanslutning>Cosmos DB.

   

3. I fönstret Skapa dataanslutning i Cosmos DB fyller du i formuläret med informationen i tabellen:

   

   Fält	Beskrivning
   Databasnamn	Välj den Azure-Data Explorer databas som du vill mata in data i.
   Namn på dataanslutning	Ange ett namn för dataanslutningen.
   Prenumeration	Välj den prenumeration som innehåller ditt Cosmos DB NoSQL-konto.
   Cosmos DB-konto	Välj det Cosmos DB-konto som du vill mata in data från.
   SQL-databas	Välj den Cosmos DB-databas som du vill mata in data från.
   SQL-container	Välj den Cosmos DB-container som du vill mata in data från.
   Tabellnamn	Ange det Azure Data Explorer tabellnamn som du vill mata in data till.
   Mappningsnamn	Du kan också ange det mappningsnamn som ska användas för dataanslutningen.

4. Du kan också göra följande under avsnittet Avancerade inställningar :

   1. Ange startdatum för händelsehämtning. Det här är den tid då anslutningsappen börjar mata in data. Om du inte anger någon tid börjar anslutningsappen mata in data från det att du skapar dataanslutningen. Det rekommenderade datumformatet är ISO 8601 UTC-standarden, som anges på följande sätt: yyyy-MM-ddTHH:mm:ss.fffffffZ.

   2. Välj Användartilldelad och välj sedan identiteten. Som standard används den systemtilldelade hanterade identiteten av anslutningen. Om det behövs kan du använda en användartilldelad identitet.

      

5. Välj Skapa för att hämta dataanslutningen.

ARM-mall

Använd följande ARM-exempelmall som grund för att skapa en egen mall för dataanslutning och distribuera den sedan i Azure Portal.

Så här konfigurerar du Cosmos DB-anslutningen:

1. Konfigurera en systemhanterad identitet för din Cosmos DB-anslutningsautentisering.

   1. I webbgränssnittet för Azure Data Explorer väljer du Fråga i den vänstra navigeringsmenyn och väljer sedan klustret eller databasen för dataanslutningen.

2. Ge dataanslutningen behörighet att komma åt ditt Cosmos DB-konto. Genom att tillhandahålla dataanslutningsåtkomst till din Cosmos DB kan den komma åt och hämta data från databasen. Du behöver klustrets huvudnamns-ID, som du hittar i Azure Portal. Mer information finns i Konfigurera hanterade identiteter för klustret.

   Anteckning

   • Följande steg tilldelar dessa roller till huvudkonto-ID:t:
     • Inbyggd dataläsare i Cosmos DB
       • Du kan inte tilldela rollen Inbyggd dataläsare i Cosmos DB med hjälp av funktionen Azure Portal rolltilldelning.
     • Cosmos DB-kontoläsarroll

   Använd något av följande alternativ för att bevilja åtkomst till ditt Cosmos DB-konto:

   • Bevilja åtkomst med hjälp av Azure CLI: Kör CLI-kommandot med hjälp av informationen i följande tabell för att ersätta platshållarna med lämpliga värden:

     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>
     

     Platshållare	Description
     <CosmosDBAccountName>	Namnet på ditt Cosmos DB-konto.
     <CosmosDBResourceGroup>	Namnet på resursgruppen som innehåller ditt Cosmos DB-konto.
     <CosmosDBAccountResourceId>	Azure-resurs-ID :t (börjar med subscriptions/) för ditt Cosmos DB-konto.
     <ClusterPrincipalId>	Huvud-ID:t för den hanterade identiteten som tilldelats klustret. Du hittar klustrets princip-ID i Azure Portal. Mer information finns i Konfigurera hanterade identiteter för klustret.

   • Bevilja åtkomst med hjälp av en ARM-mall: Distribuera följande mall i Cosmos DB-kontoresursgruppen:

     {
         "$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. Distribuera följande ARM-mall för att skapa en Cosmos DB-dataanslutning. Ersätt platshållarna med lämpliga värden.

   {
     "$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')]"
       }
     }]
   }
   

Steg 3: Testa dataanslutningen

1. Infoga följande dokument i Cosmos DB-containern:

   {
       "name":"Cousteau"
   }
   

2. Kör följande fråga i webbgränssnittet för Azure Data Explorer:

   TestTable
   

   Resultatuppsättningen bör se ut som på följande bild:

   

Anteckning

Azure Data Explorer har en aggregeringsprincip (batchbearbetning) för datainmatning i kö som är utformad för att optimera inmatningsprocessen. Standardprincipen för batchbearbetning är konfigurerad för att försegla en batch när något av följande villkor gäller för batchen: en maximal fördröjningstid på 5 minuter, total storlek på en GB eller 1 000 blobar. Därför kan det uppstå en fördröjning. Mer information finns i batchbearbetningsprincip. Om du vill minska svarstiden konfigurerar du tabellen så att den stöder strömning. Se strömningsprincip.

Överväganden

Följande överväganden gäller för Cosmos DB-ändringsflödet:

• Ändringsflödet exponerar inte borttagningshändelser .

  Cosmos DB-ändringsflödet innehåller bara nya och uppdaterade dokument. Om du behöver veta mer om borttagna dokument kan du konfigurera flödet med en mjuk markör för att markera ett Cosmos DB-dokument som borttaget. En egenskap läggs till för att uppdatera händelser som anger om ett dokument har tagits bort. Du kan sedan använda operatorn where i dina frågor för att filtrera bort dem.

  Om du till exempel mappar den borttagna egenskapen till en tabellkolumn med namnet IsDeleted kan du filtrera bort borttagna dokument med följande fråga:

  TestTable
  | where not(IsDeleted)
  

• Ändringsflödet exponerar bara den senaste uppdateringen av ett dokument.

  För att förstå konsekvenserna av det andra övervägandet undersöker du följande scenario:

  En Cosmos DB-container innehåller dokumenten A och B. Ändringarna i en egenskap som kallas foo visas i följande tabell:

  Dokument-ID	Egenskapsfoo	Händelse	Tidsstämpel för dokument (_ts)
  A	Röd	Skapa	10
  B	Blue	Skapa	20
  A	Orange	Uppdatera	30
  A	Rosa	Uppdatera	40
  B	Violett	Uppdatera	50
  A	Carmine	Uppdatera	50
  B	NeonBlue	Uppdatera	70

  API:et för ändringsflöde avsöks med jämna mellanrum av dataanslutningsappen, vanligtvis med några sekunders mellanrum. Varje avsökning innehåller ändringar som har inträffat i containern mellan anrop, men endast den senaste versionen av ändringen per dokument.

  För att illustrera problemet bör du överväga en sekvens med API-anrop med tidsstämplarna 15, 35, 55 och 75 enligt följande tabell:

  Tidsstämpel för API-anrop	Dokument-ID	Egenskapsfoo	Tidsstämpel för dokument (_ts)
  15	A	Röd	10
  35	B	Blue	20
  35	A	Orange	30
  55	B	Violett	50
  55	A	Carmine	60
  75	B	NeonBlue	70

  Om du jämför API-resultaten med listan över ändringar som gjorts i Cosmos DB-dokumentet ser du att de inte matchar. Uppdateringshändelsen för att dokumentera A, markerad i ändringstabellen vid tidsstämpel 40, visas inte i resultatet av API-anropet.

  För att förstå varför händelsen inte visas undersöker vi ändringarna i dokument A mellan API-anropen vid tidsstämplarna 35 och 55. Mellan dessa två anrop har dokument A ändrats två gånger, enligt följande:

  Dokument-ID	Egenskaps-foo	Händelse	Tidsstämpel för dokument (_ts)
  A	Rosa	Uppdatera	40
  A	Carmine	Uppdatera	50

  När API-anropet vid tidsstämpel 55 görs returnerar ändringsflödes-API:et den senaste versionen av dokumentet. I det här fallet är den senaste versionen av dokument A uppdateringen vid tidsstämpel 50, vilket är uppdateringen av egenskapen foo från Pink till Carmine.

  På grund av det här scenariot kan dataanslutningsappen missa vissa mellanliggande dokumentändringar. Vissa händelser kan till exempel missas om dataanslutningstjänsten är nere i några minuter eller om frekvensen för dokumentändringar är högre än API-avsökningsfrekvensen. Det senaste tillståndet för varje dokument samlas dock in.

• Det går inte att ta bort och återskapa en Cosmos DB-container

  Azure Data Explorer håller reda på ändringsflödet genom att kontrollera "positionen" som den befinner sig på i flödet. Detta görs med hjälp av fortsättningstoken på varje fysisk partition i containern. När en container tas bort/återskapas är dessa fortsättningstoken ogiltiga och återställs inte; du måste ta bort och återskapa dataanslutningen.

Uppskatta kostnad

Hur mycket påverkar användningen av Cosmos DB-dataanslutningen din Cosmos DB-containers användning av enheter för programbegäran (RU:er ) ?

Anslutningsappen anropar Cosmos DB Change Feed API på varje fysisk partition i containern, upp till en gång i sekunden. Följande kostnader är kopplade till dessa anrop:

Kostnad	Description
Fasta kostnader	Fasta kostnader är cirka 2 RU:er per fysisk partition varje sekund.
Rörliga kostnader	Rörliga kostnader är cirka 2 % av de RU:er som används för att skriva dokument, även om detta kan variera beroende på ditt scenario. Om du till exempel skriver 100 dokument till en Cosmos DB-container är kostnaden för att skriva dessa dokument 1 000 RU:er. Motsvarande kostnad för att använda anslutningsappen för att läsa dokumentet är ungefär 2 % kostnaden för att skriva dem, cirka 20 RU:er.

Relaterat innehåll

• Hämta de senaste versionerna av Azure Cosmos DB-dokument
• Översikt över Kusto Query Language (KQL)