Ingerir dados de exemplo formatados JSON no Azure Data Explorer

Este artigo mostra-lhe como ingerir dados formatados JSON numa base de dados do Azure Data Explorer. Irá começar com exemplos simples de JSON não processado e mapeado, continuar para JSON multi-forrado e, em seguida, abordar esquemas JSON mais complexos que contenham matrizes e dicionários. Os exemplos detalham o processo de ingestão de dados formatados JSON com Linguagem de Pesquisa Kusto (KQL), C#ou Python.

Nota

Não recomendamos a utilização de .ingest comandos de gestão em cenários de produção. Em vez disso, utilize um conector de dados ou ingera dados através de programação através de uma das bibliotecas de cliente kusto.

Pré-requisitos

• Uma conta Microsoft ou uma identidade de utilizador Microsoft Entra. Não é necessária uma subscrição do Azure.
• Um cluster e uma base de dados do Azure Data Explorer. Criar um cluster e uma base de dados.

O formato JSON

O Azure Data Explorer suporta dois formatos de ficheiro JSON:

• json: JSON separado por linhas. Cada linha nos dados de entrada tem exatamente um registo JSON. Este formato suporta a análise de comentários e propriedades com uma única citação. Para obter mais informações, veja Linhas JSON.
• multijson: JSON multi-forrado. O analisador ignora os separadores de linha e lê um registo da posição anterior até ao fim de um JSON válido.

Nota

Ao ingerir com o assistente de ingestão, o formato predefinido é multijson. O formato pode processar registos JSON multiline e matrizes de registos JSON. Quando é encontrado um erro de análise, todo o ficheiro é eliminado. Para ignorar registos JSON inválidos, selecione a opção "Ignorar erros de formato de dados", que mudará o formato para json (Linhas JSON).

Se estiver a utilizar o formato de Linha JSON (json), as linhas que não representam registos JSON válidos são ignoradas durante a análise.

Ingerir e mapear dados formatados JSON

A ingestão de dados formatados JSON requer que especifique o formato com a propriedade ingestão. A ingestão de dados JSON requer mapeamento, que mapeia uma entrada de origem JSON para a coluna de destino. Ao ingerir dados, utilize a propriedade com a IngestionMapping respetiva ingestionMappingReference propriedade de ingestão (para um mapeamento predefinido) ou a respetiva IngestionMappings propriedade. Este artigo utilizará a ingestionMappingReference propriedade ingestão, que é predefinida na tabela utilizada para ingestão. Nos exemplos abaixo, vamos começar por ingerir registos JSON como dados não processados numa única tabela de colunas. Em seguida, vamos utilizar o mapeamento para ingerir cada propriedade na coluna mapeada.

Exemplo JSON simples

O exemplo seguinte é um JSON simples, com uma estrutura plana. Os dados têm informações de temperatura e humidade, recolhidas por vários dispositivos. Cada registo é marcado com um ID e carimbo de data/hora.

{
    "timestamp": "2019-05-02 15:23:50.0369439",
    "deviceId": "2945c8aa-f13e-4c48-4473-b81440bb5ca2",
    "messageId": "7f316225-839a-4593-92b5-1812949279b3",
    "temperature": 31.0301639051317,
    "humidity": 62.0791099602725
}

Ingerir registos JSON não processados

Neste exemplo, irá ingerir registos JSON como dados não processados numa única tabela de colunas. A manipulação de dados, a utilização de consultas e a política de atualização são efetuadas após a ingestão dos dados.

KQL

Utilize Linguagem de Pesquisa Kusto para ingerir dados num formato JSON não processado.

1. Inicie sessão em https://dataexplorer.azure.com.

2. Selecione Adicionar cluster.

3. Na caixa de diálogo Adicionar cluster , introduza o URL do cluster no formulário https://<ClusterName>.<Region>.kusto.windows.net/e, em seguida, selecione Adicionar.

4. Cole o seguinte comando e selecione Executar para criar a tabela.

   .create table RawEvents (Event: dynamic)
   

   Esta consulta cria uma tabela com uma única Event coluna de um tipo de dados dinâmico .

5. Crie o mapeamento JSON.

   .create table RawEvents ingestion json mapping 'RawEventMapping' '[{"column":"Event","Properties":{"path":"$"}}]'
   

   Este comando cria um mapeamento e mapeia o caminho $ de raiz JSON para a Event coluna.

6. Ingerir dados na RawEvents tabela.

   .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"RawEventMapping"}'
   

C#

Utilize C# para ingerir dados em formato JSON não processado.

1. Crie a RawEvents tabela.

   var kustoUri = "https://<clusterName>.<region>.kusto.windows.net/";
   var connectionStringBuilder = new KustoConnectionStringBuilder(kustoUri)
   {
       FederatedSecurity = true,
       UserID = userId,
       Password = password,
       Authority = tenantId,
       InitialCatalog = databaseName
   };
   using var kustoClient = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
   var tableName = "RawEvents";
   var command = CslCommandGenerator.GenerateTableCreateCommand(
       tableName,
       new[] { Tuple.Create("Events", "System.Object") }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

2. Crie o mapeamento JSON.

   var tableMappingName = "RawEventMapping";
   command = CslCommandGenerator.GenerateTableMappingCreateCommand(
       IngestionMappingKind.Json,
       tableName,
       tableMappingName,
       new ColumnMapping[]
       {
           new() { ColumnName = "Events", Properties = new Dictionary<string, string> { { "path", "$" } } }
       }
   );
   
   await kustoClient.ExecuteControlCommandAsync(command);
   

   Este comando cria um mapeamento e mapeia o caminho $ de raiz JSON para a Event coluna.

3. Ingerir dados na RawEvents tabela.

   var ingestUri = "https://ingest-<clusterName>.<region>.kusto.windows.net/";
   
   var ingestConnectionStringBuilder = new KustoConnectionStringBuilder(ingestUri)
   {
       FederatedSecurity = true,
       UserID = userId,
       Password = password,
       Authority = tenantId,
       InitialCatalog = databaseName
   };
   using var ingestClient = KustoIngestFactory.CreateQueuedIngestClient(ingestConnectionStringBuilder);
   
   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.json,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties);
   

Nota

Os dados são agregados de acordo com a política de criação de lotes, o que resulta numa latência de alguns minutos.

Python

Utilize o Python para ingerir dados no formato JSON não processado.

1. Crie a RawEvents tabela.

   KUSTO_URI = "https://<ClusterName>.<Region>.kusto.windows.net/"
   KCSB_DATA = KustoConnectionStringBuilder.with_aad_device_authentication(KUSTO_URI, AAD_TENANT_ID)
   KUSTO_CLIENT = KustoClient(KCSB_DATA)
   TABLE = "RawEvents"
   
   CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Events: dynamic)"
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

2. Crie o mapeamento JSON.

   MAPPING = "RawEventMapping"
   CREATE_MAPPING_COMMAND = ".create table " + TABLE + " ingestion json mapping '" + MAPPING + """' '[{"column":"Event","path":"$"}]'"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. Ingerir dados na RawEvents tabela.

   INGEST_URI = "https://ingest-<ClusterName>.<Region>.kusto.windows.net/"
   KCSB_INGEST = KustoConnectionStringBuilder.with_aad_device_authentication(INGEST_URI, AAD_TENANT_ID)
   INGESTION_CLIENT = KustoIngestClient(KCSB_INGEST)
   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
   
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

   Nota

   Os dados são agregados de acordo com a política de criação de lotes, o que resulta numa latência de alguns minutos.

Ingerir registos JSON mapeados

Neste exemplo, irá ingerir dados de registos JSON. Cada propriedade JSON é mapeada para uma única coluna na tabela.

KQL

1. Crie uma nova tabela, com um esquema semelhante aos dados de entrada JSON. Vamos utilizar esta tabela para todos os seguintes exemplos e comandos de ingestão.

   .create table Events (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)
   

2. Crie o mapeamento JSON.

   .create table Events ingestion json mapping 'FlatEventMapping' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'
   

   Neste mapeamento, conforme definido pelo esquema da tabela, as timestamp entradas serão ingeridas na coluna Time como datetime tipos de dados.

3. Ingerir dados na Events tabela.

   .ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"FlatEventMapping"}'
   

   O ficheiro "simple.json" tem alguns registos JSON separados por linhas. O formato é json, e o mapeamento utilizado no comando ingestão é o FlatEventMapping que criou.

C#

1. Crie uma nova tabela, com um esquema semelhante aos dados de entrada JSON. Vamos utilizar esta tabela para todos os seguintes exemplos e comandos de ingestão.

   var tableName = "Events";
   var command = CslCommandGenerator.GenerateTableCreateCommand(
      tableName,
      new[]
      {
          Tuple.Create("Time", "System.DateTime"),
          Tuple.Create("Device", "System.String"),
          Tuple.Create("MessageId", "System.String"),
          Tuple.Create("Temperature", "System.Double"),
          Tuple.Create("Humidity", "System.Double")
      }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

2. Crie o mapeamento JSON.

   var tableMappingName = "FlatEventMapping";
   command = CslCommandGenerator.GenerateTableMappingCreateCommand(
       IngestionMappingKind.Json,
       tableName,
       tableMappingName,
       new ColumnMapping[]
       {
           new() { ColumnName = "Time", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.timestamp" } } },
           new() { ColumnName = "Device", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.deviceId" } } },
           new() { ColumnName = "MessageId", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.messageId" } } },
           new() { ColumnName = "Temperature", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.temperature" } } },
           new() { ColumnName = "Humidity", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.humidity" } } }
       }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

   Neste mapeamento, conforme definido pelo esquema da tabela, as timestamp entradas serão ingeridas na coluna Time como datetime tipos de dados.

3. Ingerir dados na Events tabela.

   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.json,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);
   

   O ficheiro "simple.json" tem alguns registos JSON separados por linhas. O formato é json, e o mapeamento utilizado no comando ingestão é o FlatEventMapping que criou.

Python

1. Crie uma nova tabela, com um esquema semelhante aos dados de entrada JSON. Vamos utilizar esta tabela para todos os seguintes exemplos e comandos de ingestão.

   TABLE = "Events"
   CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)"
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

2. Crie o mapeamento JSON.

   MAPPING = "FlatEventMapping"
   CREATE_MAPPING_COMMAND = ".create table Events ingestion json mapping '" + MAPPING + """' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. Ingerir dados na Events tabela.

   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
   
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

   O ficheiro "simple.json" tem alguns registos JSON separados por linhas. O formato é json, e o mapeamento utilizado no comando ingestão é o FlatEventMapping que criou.

Ingerir registos JSON multilindrados

Neste exemplo, irá ingerir registos JSON multilindrados. Cada propriedade JSON é mapeada para uma única coluna na tabela. O ficheiro "multilined.json" tem alguns registos JSON com avanço. O formato multijson indica para ler registos pela estrutura JSON.

KQL

Ingerir dados na Events tabela.

.ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json') with '{"format":"multijson", "ingestionMappingReference":"FlatEventMapping"}'

C#

Ingerir dados na Events tabela.

var tableMappingName = "FlatEventMapping";
var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
{
    Format = DataSourceFormat.multijson,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
};
await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);

Python

Ingerir dados na Events tabela.

MAPPING = "FlatEventMapping"
BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

Ingerir registos JSON que contêm matrizes

Os tipos de dados de matriz são uma coleção ordenada de valores. A ingestão de uma matriz JSON é feita por uma política de atualização. O JSON é ingerido tal como está numa tabela intermédia. Uma política de atualização executa uma função predefinida na RawEvents tabela, reestando os resultados para a tabela de destino. Iremos ingerir dados com a seguinte estrutura:

{
    "records":
    [
        {
            "timestamp": "2019-05-02 15:23:50.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "7f316225-839a-4593-92b5-1812949279b3",
            "temperature": 31.0301639051317,
            "humidity": 62.0791099602725
        },
        {
            "timestamp": "2019-05-02 15:23:51.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "57de2821-7581-40e4-861e-ea3bde102364",
            "temperature": 33.7529423105311,
            "humidity": 75.4787976739364
        }
    ]
}

KQL

1. Crie uma função update policy que expanda a coleção de records para que cada valor na coleção receba uma linha separada, utilizando o mv-expand operador. Vamos utilizar a tabela RawEvents como uma tabela de origem e Events como uma tabela de destino.

   .create function EventRecordsExpand() {
       RawEvents
       | mv-expand records = Event.records
       | project
           Time = todatetime(records["timestamp"]),
           Device = tostring(records["deviceId"]),
           MessageId = tostring(records["messageId"]),
           Temperature = todouble(records["temperature"]),
           Humidity = todouble(records["humidity"])
   }
   

2. O esquema recebido pela função tem de corresponder ao esquema da tabela de destino. Utilize getschema o operador para rever o esquema.

   EventRecordsExpand() | getschema
   

3. Adicione a política de atualização à tabela de destino. Esta política irá executar automaticamente a consulta em quaisquer dados recentemente ingeridos na RawEvents tabela intermédia e ingerir os resultados na Events tabela. Defina uma política de retenção zero para evitar manter a tabela intermédia.

   .alter table Events policy update @'[{"Source": "RawEvents", "Query": "EventRecordsExpand()", "IsEnabled": "True"}]'
   

4. Ingerir dados na RawEvents tabela.

   .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json') with '{"format":"multijson", "ingestionMappingReference":"RawEventMapping"}'
   

5. Reveja os dados na Events tabela.

   Events
   

C#

1. Crie uma função de atualização que expanda a coleção de records para que cada valor na coleção receba uma linha separada, utilizando o mv-expand operador. Vamos utilizar a tabela RawEvents como uma tabela de origem e Events como uma tabela de destino.

   var command = CslCommandGenerator.GenerateCreateFunctionCommand(
       "EventRecordsExpand",
       "UpdateFunctions",
       string.Empty,
       null,
       @"RawEvents
       | mv-expand records = Event
       | project
           Time = todatetime(records['timestamp']),
           Device = tostring(records['deviceId']),
           MessageId = tostring(records['messageId']),
           Temperature = todouble(records['temperature']),
           Humidity = todouble(records['humidity'])",
       ifNotExists: false
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

   Nota

   O esquema recebido pela função tem de corresponder ao esquema da tabela de destino.

2. Adicione a política de atualização à tabela de destino. Esta política irá executar automaticamente a consulta em quaisquer dados recentemente ingeridos na RawEvents tabela intermédia e ingerir os respetivos resultados na Events tabela. Defina uma política de retenção zero para evitar manter a tabela intermédia.

   command = ".alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]";
   await kustoClient.ExecuteControlCommandAsync(command);
   

3. Ingerir dados na RawEvents tabela.

   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json";
   var tableName = "RawEvents";
   var tableMappingName = "RawEventMapping";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.multijson,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties);
   

4. Reveja os dados na Events tabela.

Python

1. Crie uma função de atualização que expanda a coleção de records para que cada valor na coleção receba uma linha separada, utilizando o mv-expand operador. Vamos utilizar a tabela RawEvents como uma tabela de origem e Events como uma tabela de destino.

   CREATE_FUNCTION_COMMAND =
       '''.create function EventRecordsExpand() {
           RawEvents
           | mv-expand records = Event
           | project
               Time = todatetime(records["timestamp"]),
               Device = tostring(records["deviceId"]),
               MessageId = tostring(records["messageId"]),
               Temperature = todouble(records["temperature"]),
               Humidity = todouble(records["humidity"])
           }'''
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_FUNCTION_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

   Nota

   O esquema recebido pela função tem de corresponder ao esquema da tabela de destino.

2. Adicione a política de atualização à tabela de destino. Esta política irá executar automaticamente a consulta em quaisquer dados recentemente ingeridos na RawEvents tabela intermédia e ingerir os respetivos resultados na Events tabela. Defina uma política de retenção zero para evitar manter a tabela intermédia.

   CREATE_UPDATE_POLICY_COMMAND =
       """.alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_UPDATE_POLICY_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. Ingerir dados na RawEvents tabela.

   TABLE = "RawEvents"
   MAPPING = "RawEventMapping"
   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json'
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

4. Reveja os dados na Events tabela.

Conteúdo relacionado

• Descrição geral da ingestão de dados
• Escrever consultas