Adatok betöltése az Azure Cosmos DB-ből az Azure Data Explorer

Az Azure Data Explorer támogatja az Azure Cosmos DB for NoSql-ből történő adatbetöltést változáscsatorna használatával. A Cosmos DB változáscsatorna-adatkapcsolata egy betöltési folyamat, amely figyeli a Cosmos DB változáscsatornáját, és betölti az adatokat a Data Explorer táblába. A változáscsatorna figyeli az új és frissített dokumentumokat, de nem törli a naplókat. Az Azure Data Explorer adatbetöltésével kapcsolatos általános információkért lásd: Az Azure Data Explorer adatbetöltés áttekintése.

Minden adatkapcsolat egy adott Cosmos DB-tárolót figyel, és adatokat betölt egy adott táblába (több kapcsolat is betölthető egyetlen táblába). A betöltési módszer támogatja a streamelési betöltést (ha engedélyezve van) és az üzenetsorba történő betöltést.

Ebből a cikkből megtudhatja, hogyan állíthat be egy Cosmos DB változáscsatorna-adatkapcsolatot az adatok Azure Data Explorer való betöltéséhez a rendszer által felügyelt identitással. A kezdés előtt tekintse át a szempontokat .

Az összekötő beállításához kövesse az alábbi lépéseket:

1. lépés: Válasszon ki egy Azure Data Explorer táblát, és konfigurálja a táblaleképezést

2. lépés: Cosmos DB-adatkapcsolat létrehozása

3. lépés: Az adatkapcsolat tesztelése

Előfeltételek

• Azure-előfizetés. Hozzon létre egy ingyenes Azure-fiókot.
• Egy Azure-Data Explorer-fürt és -adatbázis. Hozzon létre egy fürtöt és egy adatbázist.
• Egy Tároló a NoSQL Cosmos DB-fiókjából.
• Ha a Cosmos DB-fiók letiltja a hálózati hozzáférést, például egy privát végpont használatával, létre kell hoznia egy felügyelt privát végpontot a Cosmos DB-fiókhoz. Ez szükséges ahhoz, hogy a fürt meghívja a változáscsatorna API-t.

1. lépés: Válasszon ki egy Azure-Data Explorer táblát, és konfigurálja a táblaleképezést

Mielőtt létrehoz egy adatkapcsolatot, hozzon létre egy táblát, amelyben tárolja a betöltött adatokat, és alkalmazza a sémának megfelelő leképezést a forrás Cosmos DB-tárolóban. Ha a forgatókönyv több mint egyszerű mezők leképezését igényli, a frissítési szabályzatokkal átalakíthatja és leképezheti a változáscsatornából betöltött adatokat.

Az alábbiakban egy Cosmos DB-tárolóban lévő elem mintaséma látható:

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

A következő lépésekkel hozzon létre egy táblát, és alkalmazza a táblaleképezést:

1. Az Azure Data Explorer webes felhasználói felületén a bal oldali navigációs menüben válassza a Lekérdezés lehetőséget, majd válassza ki azt az adatbázist, amelyben létre szeretné hozni a táblát.

2. Futtassa a következő parancsot egy TestTable nevű tábla létrehozásához.

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

3. Futtassa az alábbi parancsot a táblaleképezés létrehozásához.

   A parancs a Cosmos DB JSON-dokumentum egyéni tulajdonságait a TestTable tábla oszlopaiba képezi le az alábbiak szerint:

   Cosmos DB tulajdonság	Táblázatoszlop	Átalakítás
   id	Id	None
   név	Name	None
   _Ts	_ts	None
   _Ts	_Időbélyeg	_ts DateTimeFromUnixSeconds(UNIX másodperc) _timestamp(datetime))

   Megjegyzés

   Javasoljuk, hogy használja a következő időbélyeg-oszlopokat:

   • _ts: Ezzel az oszlopmal egyeztethet adatokat a Cosmos DB-vel.
   • _timestamp: Ezzel az oszloppal hatékony időszűrőket futtathat a Kusto-lekérdezésekben. További információ: Lekérdezési ajánlott eljárás.

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

Adatok átalakítása és leképezése frissítési szabályzatokkal

Ha a forgatókönyv több mint egyszerű mezők leképezését igényli, a frissítési szabályzatokkal átalakíthatja és leképezheti a változáscsatornából betöltött adatokat.

A frissítési szabályzatok az adatok táblázatba való betöltésekor történő átalakítására szolgálnak. Ezek a Kusto lekérdezésnyelv vannak megírva, és a betöltési folyamaton futnak. Az adatok a Cosmos DB változáscsatorna-betöltési adatainak átalakítására használhatók, például a következő esetekben:

• A dokumentumok olyan tömböket tartalmaznak, amelyek könnyebben lekérdezhetők, ha az operátorral mv-expand több sorban vannak átalakítva.
• Ki szeretné szűrni a dokumentumokat. A dokumentumokat például típus szerint szűrheti az where operátorral.
• Olyan összetett logikával rendelkezik, amelyet nem lehet táblaleképezésben ábrázolni.

A frissítési szabályzatok létrehozásával és kezelésével kapcsolatos információkért lásd: Frissítési szabályzatok áttekintése.

2. lépés: Cosmos DB-adatkapcsolat létrehozása

Az adatösszekötőt az alábbi módszerekkel hozhatja létre:

Azure Portal

1. A Azure Portal lépjen a fürt áttekintési lapjára, majd válassza az Első lépések lapot.

2. Az Adatbetöltés csempén válassza az Adatkapcsolat> létrehozásaCosmos DB lehetőséget.

   

3. A Cosmos DB Adatkapcsolat létrehozása panelen töltse ki az űrlapot a tábla adataival:

   

   Mező	Leírás
   Adatbázis neve	Válassza ki azt az Azure Data Explorer adatbázist, amelybe adatokat szeretne betöltésre.
   Adatkapcsolat neve	Adja meg az adatkapcsolat nevét.
   Előfizetés	Válassza ki a Cosmos DB NoSQL-fiókot tartalmazó előfizetést.
   Cosmos DB-fiók	Válassza ki azt a Cosmos DB-fiókot, amelyből adatokat szeretne betöltésre.
   SQL-adatbázis	Válassza ki azt a Cosmos DB-adatbázist, amelyből adatokat szeretne betöltésre.
   SQL-tároló	Válassza ki azt a Cosmos DB-tárolót, amelyből adatokat szeretne betöltésre.
   Tábla neve	Adja meg azt az Azure Data Explorer táblanevet, amelybe adatokat szeretne betöltésre.
   Leképezés neve	Igény szerint adja meg az adatkapcsolathoz használandó leképezési nevet .

4. Ha szeretné, a Speciális beállítások szakaszban tegye a következőket:

   1. Adja meg az esemény lekérésének kezdő dátumát. Ekkor kezdi meg az összekötő az adatok betöltését. Ha nem ad meg időpontot, az összekötő az adatkapcsolat létrehozásakor kezdi meg az adatok betöltését. Az ajánlott dátumformátum az ISO 8601 UTC szabvány, amely a következőképpen van megadva: yyyy-MM-ddTHH:mm:ss.fffffffZ.

   2. Válassza a Felhasználó által hozzárendelt lehetőséget, majd válassza ki az identitást. Alapértelmezés szerint a rendszer által hozzárendelt felügyelt identitást használja a kapcsolat. Szükség esetén használhat felhasználó által hozzárendelt identitást.

      

5. Válassza a Létrehozás lehetőséget az adatkapcsolat gyorsítótárazásához.

ARM-sablon

Használja az alábbi ARM-mintasablont a saját adatkapcsolati sablon létrehozásához, majd helyezze üzembe a Azure Portal.

A Cosmos DB-kapcsolat konfigurálása:

1. Konfiguráljon egy rendszer által felügyelt identitást a Cosmos DB-kapcsolat hitelesítéséhez.

   1. Az Azure Data Explorer webes felhasználói felületén válassza a bal oldali navigációs menü Lekérdezés elemét, majd válassza ki az adatkapcsolathoz tartozó fürtöt vagy adatbázist.

2. Adjon meg adatkapcsolati engedélyt a Cosmos DB-fiók eléréséhez. A Cosmos DB-hez való adatkapcsolat-hozzáférés biztosítása lehetővé teszi az adatok elérését és lekérését az adatbázisból. Szüksége lesz a fürt egyszerű azonosítójára, amelyet a Azure Portal talál. További információ: Felügyelt identitások konfigurálása a fürthöz.

   Megjegyzés

   • A következő lépések ezeket a szerepköröket rendelik hozzá az egyszerű azonosítóhoz:
     • A Cosmos DB beépített adatolvasója
       • A Cosmos DB beépített adatolvasó szerepkörét nem rendelheti hozzá a Azure Portal Szerepkör-hozzárendelés funkcióval.
     • Cosmos DB-fiókolvasó szerepkör

   A Cosmos DB-fiókhoz való hozzáférés biztosításához használja az alábbi lehetőségek egyikét:

   • Hozzáférés biztosítása az Azure CLI-vel: Futtassa a CLI-parancsot az alábbi táblázatban található információk használatával a helyőrzők megfelelő értékekre való lecseréléséhez:

     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>
     

     Helyőrző	Description
     <CosmosDBAccountName>	A Cosmos DB-fiók neve.
     <CosmosDBResourceGroup>	A Cosmos DB-fiókot tartalmazó erőforráscsoport neve.
     <CosmosDBAccountResourceId>	A Cosmos DB-fiók Azure-erőforrás-azonosítója (kezdve a következővel subscriptions/):
     <ClusterPrincipalId>	A fürthöz rendelt felügyelt identitás egyszerű azonosítója. A fürt alapvető azonosítóját a Azure Portal találja. További információ: Felügyelt identitások konfigurálása a fürthöz.

   • Hozzáférés biztosítása ARM-sablon használatával: Telepítse a következő sablont a Cosmos DB-fiók erőforráscsoportjában:

     {
         "$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. Az alábbi ARM-sablon üzembe helyezése Cosmos DB-adatkapcsolat létrehozásához. Cserélje le a helyőrzőket a megfelelő értékekre.

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

3. lépés: Az adatkapcsolat tesztelése

1. A Cosmos DB-tárolóba szúrja be a következő dokumentumot:

   {
       "name":"Cousteau"
   }
   

2. Az Azure Data Explorer webes felhasználói felületén futtassa a következő lekérdezést:

   TestTable
   

   Az eredményhalmaznak a következő képhez hasonlóan kell kinéznie:

   

Megjegyzés

Az Azure Data Explorer egy aggregációs (kötegelési) szabályzattal rendelkezik az üzenetsoros adatbetöltéshez, amelynek célja a betöltési folyamat optimalizálása. Az alapértelmezett kötegelési szabályzat úgy van konfigurálva, hogy lezárjon egy köteget, ha az alábbi feltételek egyike teljesül a kötegre vonatkozóan: legfeljebb 5 perc késleltetési idő, egy GB teljes méret vagy 1000 blob. Ezért előfordulhat, hogy késést tapasztal. További információ: Kötegelési szabályzat. A késés csökkentése érdekében konfigurálja a táblát a streamelés támogatásához. Lásd a streamelési szabályzatot.

Megfontolandó szempontok

A Cosmos DB változáscsatornájára a következő szempontok vonatkoznak:

• A változáscsatorna nem teszi közzé a törlési eseményeket.

  A Cosmos DB változáscsatornája csak új és frissített dokumentumokat tartalmaz. Ha tudnia kell a törölt dokumentumokról, konfigurálhatja a hírcsatornát egy helyreállítható jelölővel a Cosmos DB-dokumentumok töröltként való megjelöléséhez. A rendszer hozzáad egy tulajdonságot a frissítési eseményekhez, amelyek jelzik, hogy törölték-e a dokumentumot. Ezután a lekérdezések operátorával where szűrheti őket.

  Ha például leképezi a törölt tulajdonságot egy IsDeleted nevű táblaoszlopra, a következő lekérdezéssel szűrheti ki a törölt dokumentumokat:

  TestTable
  | where not(IsDeleted)
  

• A változáscsatorna csak egy dokumentum legújabb frissítését teszi elérhetővé.

  A második szempont következményeinek megértéséhez vizsgálja meg a következő forgatókönyvet:

  A Cosmos DB-tároló az A és A dokumentumokat tartalmazza. A foo nevű tulajdonság módosításai az alábbi táblázatban láthatók:

  Dokumentumazonosító	Tulajdonság foo	Esemény	Dokumentum időbélyege (_ts)
  A	Vörös	Létrehozás	10
  B	Kék	Létrehozás	20
  A	Narancssárga	Frissítés	30
  A	Rózsaszín	Frissítés	40
  B	Lila	Frissítés	50
  A	Carmine	Frissítés	50
  B	NeonBlue	Frissítés	70

  A változáscsatorna API-t az adatösszekötő rendszeres időközönként, általában néhány másodpercenként kérdezi le. Minden szavazás tartalmaz olyan módosításokat, amelyek a tárolóban történtek a hívások között, de dokumentumonként csak a legújabb módosítási verziót.

  A probléma szemléltetéséhez fontolja meg a 15-ös, 35-ös, 55-ös és 75-ös időbélyeggel rendelkező API-hívások sorozatát az alábbi táblázatban látható módon:

  API-hívás időbélyege	Dokumentumazonosító	Tulajdonság foo	Dokumentum időbélyege (_ts)
  15	A	Vörös	10
  35	B	Kék	20
  35	A	Narancssárga	30
  55	B	Lila	50
  55	A	Carmine	60
  75	B	NeonBlue	70

  Ha összehasonlítja az API-eredményeket a Cosmos DB-dokumentumban végrehajtott módosítások listájával, láthatja, hogy nem egyeznek. Az A dokumentum frissítési eseménye, amely a változástáblában a 40-edik időbélyegnél van kiemelve, nem jelenik meg az API-hívás eredményei között.

  Annak megértéséhez, hogy miért nem jelenik meg az esemény, megvizsgáljuk az A dokumentum módosításait az API-hívások között a 35-ös és az 55-ös időbélyegnél. A két hívás között az A dokumentum kétszer módosult, az alábbiak szerint:

  Dokumentumazonosító	Tulajdonság foo	Esemény	Dokumentum időbélyege (_ts)
  A	Rózsaszín	Frissítés	40
  A	Carmine	Frissítés	50

  Amikor az API-hívás az 55-ös időbélyegnél történik, a változáscsatorna API a dokumentum legújabb verzióját adja vissza. Ebben az esetben az A dokumentum legújabb verziója az 50-es időbélyegnél megadott frissítés, amely a foo tulajdonság frissítése PinkrőlCarmine-ra.

  Emiatt a forgatókönyv miatt előfordulhat, hogy az adatösszekötő kihagy néhány köztes dokumentummódosítást. Előfordulhat például, hogy bizonyos események kimaradnak, ha az adatkapcsolati szolgáltatás néhány percig nem működik, vagy ha a dokumentumváltozások gyakorisága magasabb, mint az API lekérdezési gyakorisága. A rendszer azonban rögzíti az egyes dokumentumok legújabb állapotát.

• A Cosmos DB-tárolók törlése és újra létrehozása nem támogatott

  Az Azure Data Explorer nyomon követi a változáscsatornát a csatornabeli „pozíció“ ellenőrzésével. Ez a tároló minden fizikai partícióján a folytatási token használatával történik. A tároló törlése/újbóli létrehozásakor a folytatási token érvénytelen, és nem állítható alaphelyzetbe: törölnie kell és újra létre kell hoznia az adatkapcsolatot.

Költségbecslés

Mennyire befolyásolja a Cosmos DB-adatkapcsolat használata a Cosmos DB-tároló kérelemegységek (RU-k) használatát?

Az összekötő meghívja a Cosmos DB Változáscsatorna API-t a tároló minden fizikai partícióján, akár másodpercenként egyszer is. Ezekhez a meghívásokhoz a következő költségek tartoznak:

Költségek	Description
Fix költségek	A rögzített költségek fizikai partíciónként másodpercenként körülbelül 2 kérelemegységet jelentenek.
Változó költségek	A változó költségek a dokumentumok írásához használt kérelemegységek körülbelül 2%-át képezik, bár ez a forgatókönyvtől függően változhat. Ha például 100 dokumentumot ír egy Cosmos DB-tárolóba, a dokumentumok írásának költsége 1000 kérelemegység. Az összekötőnek a dokumentum olvasására való használatának megfelelő költsége körülbelül 2%-a az írási költségeknek, körülbelül 20 kérelemegységnek.

Kapcsolódó tartalom

• Az Azure Cosmos DB-dokumentumok legújabb verzióinak lekérése
• A Kusto Query Language (KQL) áttekintése