Ingesta de datos de Azure Cosmos DB en Azure Data Explorer

Azure Data Explorer admite la ingesta de datos de Azure Cosmos DB para NoSql mediante una fuente de cambios. La conexión de datos de la alimentación de cambios de Cosmos DB es una canalización de ingestión que monitorea la alimentación de cambios de Cosmos DB e ingiere los datos en la tabla de Data Explorer. La fuente de cambios escucha para ver si hay documentos nuevos y actualizados, pero no registra eliminaciones. Para obtener información general sobre la ingesta de datos en Azure Data Explorer, consulte Introducción a la ingesta de datos de Azure Data Explorer.

Cada conexión de datos escucha un contenedor específico de Cosmos DB e ingiere datos en una tabla especificada (más de una conexión puede ingerir en una sola tabla). El método de ingesta admite la ingesta por transmisión (cuando está habilitada) y la ingesta en cola.

Los dos escenarios principales para usar la conexión de datos de fuente de cambios de Cosmos DB son:

Replicación de un contenedor de Cosmos DB con fines analíticos. Para más información, vea Obtener las versiones más recientes de los documentos de Azure Cosmos DB.
Análisis de los cambios del documento en un contenedor de Cosmos DB. Para más información, consulte Consideraciones.

En este artículo, aprenderá cómo configurar una conexión de datos de flujo de cambios de Cosmos DB para ingerir datos en Azure Data Explorer con Identidad Administrada del Sistema. Revise las consideraciones antes de empezar.

Siga estos pasos para configurar un conector:

Paso 1: Elección de una tabla de Azure Data Explorer y configuración de su asignación de tablas

Paso 2: Creación de una conexión de datos de Cosmos DB

Paso 3: Prueba de la conexión de datos

Requisitos previos

Suscripción a Azure. Cree una cuenta de Azure gratuita.
Un clúster y la base de datos de Azure Data Explorer. Cree un clúster y una base de datos.
Un contenedor de una cuenta de Cosmos DB para NoSQL.
Si la cuenta de Cosmos DB bloquea el acceso a la red, por ejemplo, mediante un punto de conexión privado, debe crear un punto de conexión privado administrado en la cuenta de Cosmos DB. Este requisito permite al clúster invocar la API de fuente de cambios.

Paso 1: Elija una tabla de Azure Data Explorer y configure su mapeo de tablas

Antes de crear una conexión de datos, cree una tabla donde almacene los datos ingeridos y aplique una asignación que coincida con el esquema en el contenedor de Cosmos DB de origen. Si el escenario requiere más de una asignación sencilla de campos, puede usar directivas de actualización para transformar y asignar datos ingeridos desde el flujo de cambios.

A continuación se muestra un esquema de ejemplo de un elemento en el contenedor de 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
}

Siga estos pasos para crear una tabla y aplicar un mapeo de tabla:

En la interfaz de usuario web de Azure Data Explorer, en el menú de navegación izquierdo, seleccione Consulta y, a continuación, seleccione la base de datos donde desea crear la tabla.

Ejecute el siguiente comando para crear una tabla denominada TestTable.

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

Ejecute el comando siguiente para crear la asignación de tablas.

El comando asigna propiedades personalizadas de un documento JSON de Cosmos DB a columnas de la tabla TestTable, como se indica a continuación:

Propiedad de Cosmos DB	Columna de tabla	Transformación
id	Id.	None
name	Nombre	None
_ts	_ts	Ninguno
_ts	_timestamp	Usa `DateTimeFromUnixSeconds` para transformar_ts (segundos de UNIX) a _timestamp (`datetime`)

Nota:

Use las siguientes columnas de marca de tiempo:

_ts: use esta columna para conciliar los datos con Cosmos DB.
_timestamp: use esta columna para ejecutar filtros de tiempo eficaces en las consultas de Kusto. Para más información, consulte Procedimientos recomendados sobre las consultas.

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

Transformación y mapeo de datos con directivas de actualización

Si su escenario requiere más que una asignación sencilla de campos, use políticas de actualización para transformar y asignar datos ingeridos desde su registro de cambios.

Las directivas de actualización transforman los datos durante su ingesta en tu tabla. Escríbalos en Kusto Query Language y ejecútelos en la canalización de ingesta. Úselos para transformar datos de una ingesta de fuente de cambios de Cosmos DB, como en los escenarios siguientes:

Los documentos contienen matrices que son más fáciles de consultar si se transforman en varias filas mediante el mv-expand operador .
Desea filtrar los documentos. Por ejemplo, puede filtrar los documentos por tipo mediante el where operador .
Usted tiene una lógica compleja que no se puede representar en un mapeo de tablas.

Para obtener información sobre cómo crear y administrar directivas de actualización, consulte Introducción a la directiva de actualización.

Paso 2: Creación de una conexión de datos de Cosmos DB

Use los métodos siguientes para crear el conector de datos:

Azure Portal
Plantilla ARM

En Azure Portal, vaya a la página de información general del clúster y seleccione la pestaña Introducción.
En el icono Ingesta de datos, seleccione Crear conexión de datos>Cosmos DB.

En el panel Crear conexión de datos de Cosmos DB, rellene el formulario con la información de la tabla siguiente:

Captura de pantalla del panel de conexión de datos, en la que se muestran los campos de formulario con valores.

Campo	Descripción
Nombre de la base de datos	Elija la base de datos de Azure Data Explorer en la que desea ingerir datos.
Nombre de la conexión de datos	Especifique un nombre para la conexión de datos.
Suscripción	Seleccione la suscripción que contiene la cuenta de Cosmos DB NoSQL.
Cuenta de Cosmos DB	Elija la cuenta de Cosmos DB desde la que desea ingerir datos.
Base de datos SQL	Elija la base de datos de Cosmos DB desde la que desea ingerir datos.
Contenedor de SQL	Elija el contenedor de Cosmos DB desde el que desea ingerir datos.
Nombre de la tabla	Especifique el nombre de la tabla de Azure Data Explorer en la que desea ingerir datos.
Nombre de mapeo	Opcionalmente, especifique el nombre de asignación que se va a usar para la conexión de datos.

Opcionalmente, en la sección Configuración avanzada , escriba la siguiente información:
1. Especifique la fecha de inicio de recuperación de eventos. Este valor es la hora desde la que el conector comienza a ingerir datos. Si no especifica una hora, el conector comienza a ingerir datos desde el momento en que crea la conexión de datos. El formato de fecha recomendado es el estándar UTC ISO 8601, especificado de la siguiente manera: yyyy-MM-ddTHH:mm:ss.fffffffZ.
2. Seleccione Asignado por el usuario y, a continuación, seleccione la identidad. De forma predeterminada, la conexión usa la identidad administrada asignada por el sistema . Si es necesario, puede usar una identidad asignada por el usuario.
Seleccione Crear para crear la conexión de datos.

Use la siguiente plantilla de ARM de ejemplo como base para crear su propia plantilla de conexión de datos y, a continuación, implementarla en Azure Portal.

Para configurar la conexión de Cosmos DB:

Configure una identidad administrada del sistema para autenticar la conexión con Cosmos DB.
1. En la interfaz de usuario web de Azure Data Explorer, seleccione Consulta en el menú de navegación izquierdo y, a continuación, seleccione el clúster o la base de datos para la conexión de datos.

Conceda el permiso de conexión de datos para acceder a la cuenta de Cosmos DB. Al proporcionar el acceso a la conexión de datos a Cosmos DB, puede acceder a los datos de la base de datos y recuperarlos. Necesita el identificador principal de su clúster, que puede encontrar en el portal de Azure. Para más información, consulte Configuración de identidades administradas para el clúster.

Nota:

Los pasos siguientes asignan estos roles al identificador de entidad principal.
- Lector de datos integrado de Cosmos DB
  - No se puede asignar el rol Lector de datos integrado de Cosmos DB mediante la característica Asignación de roles de Azure Portal.
- Rol de lector de cuentas de Cosmos DB

Use una de las siguientes opciones para conceder acceso a la cuenta de Cosmos DB:

Concesión de acceso mediante la CLI de Azure: ejecute el comando de la CLI mediante la información de la tabla siguiente para reemplazar los marcadores de posición por los valores adecuados:

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>

Marcador de posición	Descripción
<CosmosDBAccountName>	Nombre de la cuenta de Cosmos DB.
<CosmosDBResourceGroup>	Nombre del grupo de recursos que contiene la cuenta de Cosmos DB.
<CosmosDBAccountResourceId>	ID de recurso de Azure (a partir de `subscriptions/`) de la cuenta de Cosmos DB.
<ClusterPrincipalId>	Identificador principal de la identidad administrada asignada a su clúster. Puede encontrar el ID principal del clúster en el portal de Azure. Para más información, consulte Configuración de identidades administradas para el clúster.

Concesión de acceso mediante una plantilla de ARM: implemente la plantilla siguiente en el grupo de recursos de la cuenta de 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')]"
            }
        }
    ]
}

Implemente la siguiente plantilla de ARM para crear una conexión de datos de Cosmos DB. Reemplazar los marcadores de posición por los valores adecuados.

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

Paso 3: Prueba de la conexión de datos

En el contenedor de Cosmos DB, inserte el siguiente documento:
```
{
    "name":"Cousteau"
}
```
En la interfaz de usuario web de Azure Data Explorer, ejecute la consulta siguiente:
```
TestTable
```
El conjunto de resultados debe tener un aspecto similar al de la siguiente imagen:

Nota:

Azure Data Explorer usa una directiva de agregación (procesamiento por lotes) para la ingesta de datos en cola que optimiza el proceso de ingesta. La directiva de procesamiento por lotes predeterminada sella un lote cuando se cumple una de las condiciones siguientes para el lote: un tiempo de retraso máximo de cinco minutos, un tamaño total de 1 GB o 1000 blobs. Por lo tanto, podría experimentar latencia. Para obtener más información, consulte la directiva de procesamiento por lotes. Para reducir la latencia, configure la tabla para admitir el streaming. Consulte la directiva de streaming.

Consideraciones

Las consideraciones siguientes se aplican a la fuente de cambios de Cosmos DB:

El feed de cambios no expone eventos de eliminación.

La fuente de cambios de Cosmos DB solo incluye documentos nuevos y actualizados. Si necesita información sobre los documentos eliminados, puede configurar la fuente para que use un marcador suave para marcar un documento de Cosmos DB como eliminado. Se agrega una propiedad para actualizar eventos que indican si se elimina un documento. Después, puede usar el operador where en las consultas para filtrarlos.

Por ejemplo, si asigna la propiedad eliminada a una columna de tabla denominada IsDeleted, puede filtrar los documentos eliminados con la consulta siguiente:
```
TestTable
| where not(IsDeleted)
```

La fuente de cambios solo muestra la última actualización de un documento.

Para comprender la ramificación de la segunda consideración, examine el siguiente escenario:

Un contenedor de Cosmos DB contiene documentos A y B. Los cambios realizados en una propiedad denominada foo se muestran en la tabla siguiente:

Id. de documento	Propiedad foo	Evento	Marca de tiempo del documento (_ts)
A	Rojo	Creación	10
B	Azul	Creación	20
A	Naranja	Actualizar	30
A	Rosa	Actualizar	40
B	Violeta	Actualizar	50
A	Carmine	Actualizar	50
B	NeonBlue	Actualizar	70

El conector de datos sondea la API de flujo de cambios a intervalos regulares, normalmente cada pocos segundos. Cada consulta contiene los cambios que se produjeron en el contenedor entre las llamadas, pero solo la versión más reciente del cambio por documento.

Para ilustrar el problema, considere una secuencia de llamadas API con marcas de tiempo 15, 35, 55y 75 como se muestra en la tabla siguiente:

Marca de tiempo de llamada API	Id. de documento	Propiedad foo	Marca de tiempo del documento (_ts)
15	A	Rojo	10
35	B	Azul	20
35	A	Naranja	30
55	B	Violeta	50
55	A	Carmine	60
75	B	NeonBlue	70

Al comparar los resultados de la API con la lista de cambios realizados en el documento de Cosmos DB, observará que no coinciden. El evento de actualización al documento A, resaltado en la tabla de cambios en la marca de tiempo 40, no aparece en los resultados de la llamada a la API.

Para comprender por qué el evento no aparece, examine los cambios en el documento A entre las llamadas API a las marcas de tiempo 35 y 55. Entre estas dos llamadas, el documento A cambió dos veces, como se indica a continuación:

Id. de documento	Propiedad foo	Evento	Marca de tiempo del documento (_ts)
A	Rosa	Actualizar	40
A	Carmine	Actualizar	50

Cuando se realiza la llamada API a la marca de tiempo de 55, la API de seguimiento de cambios devuelve la versión más reciente del documento. En este caso, la versión más reciente del documento A es la revisión en la marca de tiempo 50, que representa el cambio de la propiedad foo de Pink a Carmine.

Debido a este escenario, es posible que el conector de datos pierda algunos cambios intermedios en los documentos. Por ejemplo, el conector de datos podría perder algunos eventos si el servicio de conexión de datos está inactivo durante unos minutos o si la frecuencia de cambios en el documento es mayor que la frecuencia de sondeo de la API. Sin embargo, se captura el estado más reciente de cada documento.

No se admite la eliminación y recreación de un contenedor de Cosmos DB.

Azure Data Explorer realiza un seguimiento del flujo de cambios realizando puntos de control de la "posición" en la que se encuentra en el flujo. Este proceso usa un token de continuación en cada partición física del contenedor. Al eliminar y volver a crear un contenedor, el token de continuación deja de ser válido y no se restablece. En este caso, debe eliminar y volver a crear la conexión de datos.

Estimación del costo

¿Cuánto afecta el uso de la conexión de datos de Cosmos DB al uso de unidades de solicitud (RU) del contenedor de Cosmos DB?

El conector invoca la API de fuente de cambios de Cosmos DB en cada partición física del contenedor, hasta una vez por segundo. Los siguientes costos están asociados a estas invocaciones:

Costos	Descripción
Costos fijos	Los costos fijos son aproximadamente dos RU por partición física cada segundo.
Costos variables	Los costos variables son aproximadamente 2% de las RU que se usan para escribir documentos, aunque este valor puede variar en función de su escenario. Por ejemplo, si escribe 100 documentos en un contenedor de Cosmos DB, el costo de escribir esos documentos es de 1000 RU. El costo correspondiente para usar el conector para leer los documentos es de aproximadamente el 2% del costo de escribirlos, aproximadamente 20 RUs.

Costos

Descripción

Costos fijos

Los costos fijos son aproximadamente dos RU por partición física cada segundo.

Costos variables

Los costos variables son aproximadamente 2% de las RU que se usan para escribir documentos, aunque este valor puede variar en función de su escenario. Por ejemplo, si escribe 100 documentos en un contenedor de Cosmos DB, el costo de escribir esos documentos es de 1000 RU. El costo correspondiente para usar el conector para leer los documentos es de aproximadamente el 2% del costo de escribirlos, aproximadamente 20 RUs.

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-02-12

Compartir a través de

Ingesta de datos de Azure Cosmos DB en Azure Data Explorer

Requisitos previos

Paso 1: Elija una tabla de Azure Data Explorer y configure su mapeo de tablas

Transformación y mapeo de datos con directivas de actualización

Paso 2: Creación de una conexión de datos de Cosmos DB

Paso 3: Prueba de la conexión de datos

Consideraciones

Estimación del costo

Contenido relacionado

Comentarios

Recursos adicionales