Compartir a través de

Ingesta de datos de ejemplo con formato JSON en Azure Data Explorer

  • Artículo

En este artículo, se muestra cómo ingerir datos con formato JSON en una base de datos de Azure Data Explorer. Comenzará con ejemplos sencillos de JSON asignado y sin formato, continuará con JSON multilínea y, a continuación, abordará esquemas JSON más complejos que contengan matrices y diccionarios. En los ejemplos se detalla el proceso de ingesta de datos con formato JSON mediante Lenguaje de consulta Kusto (KQL), C#o Python.

Nota:

No se recomienda usar .ingest comandos de administración en escenarios de producción. En su lugar, use un conector de datos o ingiera datos mediante programación mediante una de las bibliotecas cliente de Kusto.

Requisitos previos

  • Una cuenta de Microsoft o una identidad de usuario de Microsoft Entra. No se necesita una suscripción a Azure.
  • Un clúster y la base de datos de Azure Data Explorer. Cree un clúster y una base de datos.

El formato JSON

Azure Data Explorer admite dos formatos de archivo JSON:

  • json: JSON separado por líneas. Cada línea de los datos de entrada tiene exactamente un registro JSON. Este formato admite el análisis de comentarios y propiedades entre comillas únicas. Para más información, consulte JSON Lines.
  • multijson: JSON de varias líneas. El analizador omite los separadores de línea y lee un registro desde la posición anterior hasta el final de un JSON válido.

Nota:

Al ingerir mediante la experiencia de obtención de datos, el formato predeterminado es multijson. El formato puede controlar registros JSON de varias líneas y matrices de registros JSON. Cuando se produce un error de análisis, se descarta todo el archivo. Para omitir los registros JSON no válidos, seleccione la opción "Omitir errores de formato de datos". Esto cambiará el formato a json (líneas JSON).

Si usa el formato de línea JSON (json), las líneas que no representan registros JSON válidos se omiten durante el análisis.

Ingesta y asignación de datos con formato JSON

La ingesta de datos con formato JSON requiere que especifique el formato mediante la propiedad de ingesta. La ingesta de datos JSON requiere la asignación, que asigna una entrada de origen JSON a su columna de destino. Al ingerir de datos, use la propiedad IngestionMapping con su propiedad de ingesta ingestionMappingReference (para una asignación predefinida) o su propiedad IngestionMappings. En este artículo se usará la propiedad de ingesta ingestionMappingReference, que está predefinida en la tabla que se utiliza para la ingesta. En los ejemplos siguientes, comenzaremos mediante la ingesta de registros JSON como datos sin procesar en una tabla de una sola columna. A continuación, usaremos la asignación para ingerir cada propiedad en su columna asignada.

Ejemplo de JSON simple

El siguiente ejemplo es un JSON simple, con una estructura plana. Los datos tienen información sobre la temperatura y la humedad, recopilados por varios dispositivos. Cada registro se marca con un identificador y una marca de tiempo.

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

Ingesta de registros JSON sin procesar

En este ejemplo, se ingieren registros JSON como datos sin procesar en una tabla de una sola columna. La manipulación de datos, mediante consultas, y la directiva de actualización se realizan una vez que se ingieren los datos.

Use Lenguaje de consulta Kusto para ingerir datos en un formato JSON sin procesar.

  1. Inicie sesión en https://dataexplorer.azure.com.

  2. Seleccione Agregar clúster.

  3. En el cuadro de diálogo Agregar clúster, escriba la URL del clúster con el formato https://<ClusterName>.<Region>.kusto.windows.net/ y después haga clic en Agregar.

  4. Pegue el siguiente comando y seleccione Ejecutar para crear la tabla.

    .create table RawEvents (Event: dynamic)

    Esta consulta crea una tabla con una sola columna Event de un tipo de datos dinámico.

  5. Cree la asignación de JSON.

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

    Este comando crea una asignación y asigna la ruta de acceso raíz de JSON $ a la columna Event.

  6. Ingiera los datos en la tabla RawEvents.

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

Ingesta de registros JSON asignados

En este ejemplo, se ingieren datos de registros JSON. Cada propiedad JSON se asigna a una sola columna de la tabla.

  1. Cree una tabla con un esquema similar a los datos de entrada JSON. Usaremos esta tabla en todos los ejemplos y comandos de ingesta siguientes.

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

  2. Cree la asignación de 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"}}]'

    En esta asignación, tal y como se define en el esquema de tabla, las entradas timestamp se ingerirán en la columna Time como tipos de datos datetime.

  3. Ingiera los datos en la tabla Events.

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

    El archivo "simple.JSON" tiene algunos registros JSON separados por líneas. El formato es json, y la asignación utilizada en el comando de ingesta es el elemento FlatEventMapping que creó.

Ingesta de registros JSON multilínea

En este ejemplo, se ingieren registros JSON multilínea. Cada propiedad JSON se asigna a una sola columna de la tabla. El archivo "multilined.json" tiene algunos registros JSON con sangría. El formato multijson indica que la estructura JSON lee los registros.

Ingiera los datos en la tabla Events.

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

Ingesta de registros JSON que contienen matrices

Los tipos de datos de matriz son una colección ordenada de valores. La ingesta de una matriz JSON se realiza mediante una directiva de actualización. El JSON se ingiere tal cual en una tabla intermedia. Una directiva de actualización ejecuta una función predefinida en la tabla RawEvents, de forma que se vuelven a ingerir los resultados en la tabla de destino. Los datos se ingerirán con la siguiente estructura:

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

  1. Cree una función update policy que expanda la recopilación de datos de records para que cada valor de la recopilación reciba una fila independiente, mediante el operador mv-expand. Se usará la tabla RawEvents como tabla de origen y Events como tabla 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. El esquema que recibe la función debe coincidir con el esquema de la tabla de destino. Utilice el operador getschema para revisar el esquema.

    EventRecordsExpand() | getschema

  3. Agregue la directiva de actualización a la tabla de destino. Esta directiva ejecuta automáticamente la consulta en todos los datos recién ingeridos en la tabla intermedia RawEvents e ingiere los resultados en la tabla Events. Defina una directiva de retención cero para evitar que se conserve la tabla intermedia.

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

  4. Ingiera los datos en la tabla RawEvents.

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

  5. Revise los datos de la tabla Events.

    Events

Contenido relacionado