Ingerir dados do Azure Cosmos DB para o Azure Data Explorer

Artigo
06/15/2023

O Azure Data Explorer suporta a ingestão de dados do Azure Cosmos DB para NoSql com um feed de alterações. A ligação de dados do feed de alterações do Cosmos DB é um pipeline de ingestão que escuta o feed de alterações do Cosmos DB e ingere os dados na sua tabela de Data Explorer. O feed de alterações escuta documentos novos e atualizados, mas não regista eliminações. Para obter informações gerais sobre a ingestão de dados no Azure Data Explorer, veja Descrição geral da ingestão de dados do Azure Data Explorer.

Cada ligação de dados escuta um contentor específico do Cosmos DB e ingere dados numa tabela especificada (mais do que uma ligação pode ingerir numa única tabela). O método de ingestão suporta a ingestão de transmissão em fluxo (quando ativada) e a ingestão em fila.

Neste artigo, irá aprender a configurar uma ligação de dados de feed de alterações do Cosmos DB para ingerir dados no Azure Data Explorer com a Identidade Gerida do Sistema. Reveja as considerações antes de começar.

Utilize os seguintes passos para configurar um conector:

Passo 1: escolher uma tabela do Azure Data Explorer e configurar o mapeamento da tabela

Passo 2: Criar uma ligação de dados do Cosmos DB

Passo 3: testar a ligação de dados

Pré-requisitos

Uma subscrição do Azure. Crie uma conta gratuita do Azure.
Um cluster e uma base de dados do Azure Data Explorer. Criar um cluster e uma base de dados.
Um contentor de uma conta do Cosmos DB para NoSQL.
Se a sua conta do Cosmos DB bloquear o acesso à rede, por exemplo, com um ponto final privado, tem de criar um ponto final privado gerido para a conta do Cosmos DB. Isto é necessário para que o cluster invoque a API do feed de alterações.

Passo 1: escolher uma tabela do Azure Data Explorer e configurar o mapeamento da tabela

Antes de criar uma ligação de dados, crie uma tabela onde irá armazenar os dados ingeridos e aplicar um mapeamento que corresponda ao esquema no contentor do Cosmos DB de origem. Se o seu cenário exigir mais do que um mapeamento simples de campos, pode utilizar políticas de atualização para transformar e mapear dados ingeridos a partir do feed de alterações.

O seguinte mostra um esquema de exemplo de um item no contentor do 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
}

Utilize os seguintes passos para criar uma tabela e aplicar um mapeamento de tabela:

Na IU do Azure Data Explorer Web, no menu de navegação esquerdo, selecione Consulta e, em seguida, selecione a base de dados onde pretende criar a tabela.

Execute o seguinte comando para criar uma tabela denominada TestTable.

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

Execute o seguinte comando para criar o mapeamento da tabela.

O comando mapeia propriedades personalizadas de um documento JSON do Cosmos DB para colunas na tabela TestTable , da seguinte forma:

Propriedade cosmos DB	Coluna de tabela	Transformação
id	Id	Nenhuma
nome	Name	Nenhuma
_ts	_ts	Nenhuma
_ts	_timestamp	Utiliza `DateTimeFromUnixSeconds` para transformar_ts (segundos UNIX) para _timestamp (`datetime`))

Nota

Recomendamos que utilize as seguintes colunas de carimbo de data/hora:

_ts: utilize esta coluna para reconciliar dados com o Cosmos DB.
_timestamp: utilize esta coluna para executar filtros de tempo eficientes nas suas consultas kusto. Para obter mais informações, veja Melhores práticas de consulta.

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

Transformar e mapear dados com políticas de atualização

Se o seu cenário exigir mais do que um mapeamento simples de campos, pode utilizar políticas de atualização para transformar e mapear dados ingeridos a partir do feed de alterações.

As políticas de atualização são uma forma de transformar dados à medida que são ingeridos na sua tabela. São escritas em Linguagem de Pesquisa Kusto e executadas no pipeline de ingestão. Podem ser utilizados para transformar dados de uma ingestão de feeds de alterações do Cosmos DB, como nos seguintes cenários:

Os seus documentos contêm matrizes que seriam mais fáceis de consultar se fossem transformados em múltiplas linhas com o mv-expand operador.
Quer filtrar documentos. Por exemplo, pode filtrar documentos por tipo com o where operador.
Tem uma lógica complexa que não pode ser representada num mapeamento de tabelas.

Para obter informações sobre como criar e gerir políticas de atualização, veja Descrição geral da política de atualização.

Passo 2: Criar uma ligação de dados do Cosmos DB

Pode utilizar os seguintes métodos para criar o conector de dados:

Portal do Azure
Modelo ARM

No portal do Azure, aceda à página de descrição geral do cluster e, em seguida, selecione o separador Introdução.
No mosaico Ingestão de dados, selecione Criar ligação> de dadosCosmos DB.

No painel Criar ligação de dados do Cosmos DB, preencha o formulário com as informações na tabela:

Captura de ecrã do painel de ligação de dados a mostrar os campos de formulário com valores.

Campo	Descrição
Nome da base de dados	Selecione a base de dados do Azure Data Explorer na qual pretende ingerir dados.
Nome da ligação de dados	Especifique um nome para a ligação de dados.
Subscrição	Selecione a subscrição que contém a sua conta NoSQL do Cosmos DB.
Conta do Cosmos DB	Selecione a conta do Cosmos DB a partir da qual pretende ingerir dados.
Base de dados SQL	Escolha a base de dados do Cosmos DB a partir da qual pretende ingerir dados.
Contentor SQL	Escolha o contentor do Cosmos DB a partir do qual pretende ingerir dados.
Nome da tabela	Especifique o nome da tabela Data Explorer do Azure ao qual pretende ingerir dados.
Nome do mapeamento	Opcionalmente, especifique o nome de mapeamento a utilizar para a ligação de dados.

Opcionalmente, na secção Definições avançadas , faça o seguinte:
1. Especifique a data de início da obtenção de eventos. Este é o momento a partir do qual o conector começará a ingerir dados. Se não especificar uma hora, o conector começará a ingerir dados a partir do momento em que criar a ligação de dados. O formato de data recomendado é a norma ISO 8601 UTC, especificada da seguinte forma: yyyy-MM-ddTHH:mm:ss.fffffffZ.
2. Selecione Atribuído pelo utilizador e, em seguida, selecione a identidade. Por Predefinição, a identidade gerida atribuída pelo sistema é utilizada pela ligação. Se necessário, pode utilizar uma identidade atribuída pelo utilizador .
Selecione Criar para colocar a ligação de dados em caixa.

Utilize o seguinte modelo arm de exemplo como base para criar o seu próprio modelo de ligação de dados e, em seguida, implemente-o no portal do Azure.

Para configurar a ligação do Cosmos DB:

Configure uma Identidade Gerida do Sistema para a autenticação de ligação do Cosmos DB.
1. Na IU do Azure Data Explorer Web, selecione Consulta no menu de navegação esquerdo e, em seguida, selecione o cluster ou a base de dados para a ligação de dados.

Conceda a permissão de ligação de dados para aceder à sua conta do Cosmos DB. Fornecer o acesso à ligação de dados ao Cosmos DB permite-lhe aceder e obter dados da sua base de dados. Precisará do ID principal do cluster, que pode encontrar no portal do Azure. Para obter mais informações, veja Configurar identidades geridas para o cluster.

Nota

Os passos seguintes atribuem estas funções ao ID principal:
- Leitor de Dados Incorporado do Cosmos DB
  - Não pode atribuir a função leitor de dados incorporado do Cosmos DB com a funcionalidade portal do Azure Atribuição de Funções.
- Função de Leitor de Conta do Cosmos DB

Utilize uma das seguintes opções para conceder acesso à sua conta do Cosmos DB:

Conceder acesso com a CLI do Azure: execute o comando da CLI com as informações na tabela seguinte para substituir marcadores de posição por valores adequados:

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 posição	Description
<CosmosDBAccountName>	O nome da sua conta do Cosmos DB.
<CosmosDBResourceGroup>	O nome do grupo de recursos que contém a sua conta do Cosmos DB.
<CosmosDBAccountResourceId>	O ID de recurso do Azure (a começar com `subscriptions/`) da sua conta do Cosmos DB.
<ClusterPrincipalId>	O ID principal da identidade gerida atribuída ao cluster. Pode encontrar o ID principal do cluster no portal do Azure. Para obter mais informações, veja Configurar identidades geridas para o cluster.

Conceder acesso com um Modelo do ARM: implemente o seguinte modelo no grupo de recursos da conta do 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 o seguinte modelo arm para criar uma ligação de dados do Cosmos DB. Substitua os marcadores de posição pelos valores adequados.

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

Passo 3: testar a ligação de dados

No contentor do Cosmos DB, insira o seguinte documento:
```
{
    "name":"Cousteau"
}
```
Na IU do Azure Data Explorer Web, execute a seguinte consulta:
```
TestTable
```
O conjunto de resultados deve ter o seguinte aspeto:

Nota

O Azure Data Explorer tem uma política de agregação (batching) para ingestão de dados em fila concebida para otimizar o processo de ingestão. A política de criação de lotes predefinida é configurada para selar um lote quando uma das seguintes condições for verdadeira para o lote: um tempo de atraso máximo de 5 minutos, tamanho total de um GB ou 1000 blobs. Por conseguinte, poderá ter uma latência. Para obter mais informações, veja Política de criação de lotes. Para reduzir a latência, configure a tabela para suportar a transmissão em fluxo. Veja a política de transmissão em fluxo.

Considerações

As seguintes considerações aplicam-se ao feed de alterações do Cosmos DB:

O feed de alterações não expõe eventos de eliminação .

O feed de alterações do Cosmos DB inclui apenas documentos novos e atualizados. Se precisar de saber mais sobre documentos eliminados, pode configurar o feed com um marcador suave para marcar um documento do Cosmos DB como eliminado. É adicionada uma propriedade para atualizar eventos que indicam se um documento foi eliminado. Em seguida, pode utilizar o where operador nas suas consultas para as filtrar.

Por exemplo, se mapear a propriedade eliminada para uma coluna de tabela denominada IsDeleted, pode filtrar documentos eliminados com a seguinte consulta:
```
TestTable
| where not(IsDeleted)
```

O feed de alterações expõe apenas a atualização mais recente de um documento.

Para compreender a ramificação da segunda consideração, examine o seguinte cenário:

Um contentor do Cosmos DB contém os documentos A e B. As alterações a uma propriedade chamada foo são apresentadas na seguinte tabela:

ID do Documento	Foo de propriedade	Evento	Carimbo de data/hora do documento (_ts)
A	Vermelho	Criação	10
N	Blue	Criação	20
A	Laranja	Atualizar	30
A	Cor-de-rosa	Atualizar	40
B	Violeta	Atualizar	50
A	Carmine	Atualizar	50
B	NeonBlue	Atualizar	70

A API do feed de alterações é consultada pelo conector de dados em intervalos regulares, normalmente a cada poucos segundos. Cada inquérito contém alterações que ocorreram no contentor entre chamadas, mas apenas a versão mais recente da alteração por documento.

Para ilustrar o problema, considere uma sequência de chamadas à API com carimbos de data/hora 15, 35, 55 e 75 , conforme mostrado na tabela seguinte:

Carimbo de Data/Hora da Chamada da API	ID do Documento	Foo de propriedade	Carimbo de data/hora do documento (_ts)
15	A	Vermelho	10
35	B	Blue	20
35	A	Laranja	30
55	B	Violeta	50
55	A	Carmine	60
75	B	Azul Néon	70

Ao comparar os resultados da API com a lista de alterações efetuadas no documento do Cosmos DB, irá reparar que não correspondem. O evento de atualização para o documento A, realçado na tabela de alterações no carimbo de data/hora 40, não aparece nos resultados da chamada à API.

Para compreender por que motivo o evento não aparece, vamos examinar as alterações ao documento A entre as chamadas à API nos carimbos de data/hora 35 e 55. Entre estas duas chamadas, o documento A foi alterado duas vezes, da seguinte forma:

ID do Documento	Foo de propriedade	Evento	Carimbo de data/hora do documento (_ts)
A	Cor-de-rosa	Atualizar	40
A	Carmine	Atualizar	50

Quando a chamada à API no carimbo de data/hora 55 é efetuada, a API do feed de alterações devolve a versão mais recente do documento. Neste caso, a versão mais recente do documento A é a atualização no carimbo de data/hora 50, que é a atualização para a propriedade foo de Rosa para Carmine.

Devido a este cenário, o conector de dados pode perder algumas alterações de documento intermédias. Por exemplo, alguns eventos podem ser perdidos se o serviço de ligação de dados estiver inativo durante alguns minutos ou se a frequência das alterações do documento for superior à frequência de consulta da API. No entanto, é capturado o estado mais recente de cada documento.

A eliminação e recriação de um contentor do Cosmos DB não é suportada

O Azure Data Explorer controla o feed de alterações ao definir o ponto de verificação da "posição" em que se encontra no feed. Tal é feito com o token de continuação em cada partição física do contentor. Quando um contentor é eliminado/recriado, esses tokens de continuação são inválidos e não são repostos: tem de eliminar e recriar a ligação de dados.

Fazer uma estimativa dos custos

Quanto é que a utilização da ligação de dados do Cosmos DB afeta a utilização das Unidades de Pedido (RUs) do contentor do Cosmos DB?

O conector invoca a API do Feed de Alterações do Cosmos DB em cada partição física do contentor, até uma vez por segundo. Os seguintes custos estão associados a estas invocações:

Custo	Description
Custos fixos	Os custos fixos são cerca de 2 RUs por partição física a cada segundo.
Custos variáveis	Os custos variáveis são cerca de 2% das RUs utilizadas para escrever documentos, embora isto possa variar consoante o seu cenário. Por exemplo, se escrever 100 documentos num contentor do Cosmos DB, o custo de escrever esses documentos é de 1000 RUs. O custo correspondente para utilizar o conector para ler esses documentos é cerca de 2% do custo de escrita, aproximadamente 20 RUs.

Partilhar via

Ingerir dados do Azure Cosmos DB para o Azure Data Explorer

Pré-requisitos

Passo 1: escolher uma tabela do Azure Data Explorer e configurar o mapeamento da tabela

Transformar e mapear dados com políticas de atualização

Passo 2: Criar uma ligação de dados do Cosmos DB

Passo 3: testar a ligação de dados

Considerações

Fazer uma estimativa dos custos

Comentários

Recursos adicionais

Partilhar via

Ingerir dados do Azure Cosmos DB para o Azure Data Explorer

Pré-requisitos

Passo 1: escolher uma tabela do Azure Data Explorer e configurar o mapeamento da tabela

Transformar e mapear dados com políticas de atualização

Passo 2: Criar uma ligação de dados do Cosmos DB

Passo 3: testar a ligação de dados

Considerações

Fazer uma estimativa dos custos

Conteúdo relacionado

Comentários

Recursos adicionais