Ingérer des exemples de données au format JSON dans Azure Data Explorer

Article
10/15/2023

Cet article vous montre comment ingérer des données au format JSON dans une base de données Azure Data Explorer. Vous commencerez par des exemples simples de données JSON brutes et mappées, puis vous passerez à des données JSON multilignes, et enfin à des schémas JSON plus complexes contenant des tableaux et des dictionnaires. Les exemples détaillent le processus d’ingestion de données au format JSON à l’aide de Langage de requête Kusto (KQL), C# ou Python.

Notes

Nous vous déconseillons d’utiliser .ingest des commandes de gestion dans les scénarios de production. Utilisez plutôt un connecteur de données ou ingérer des données par programmation à l’aide de l’une des bibliothèques clientes Kusto.

Prérequis

Un compte Microsoft ou une identité d’utilisateur Microsoft Entra. Un abonnement Azure n’est pas requis.
Un cluster et une base de données Azure Data Explorer. Créez un cluster et une base de données.

Le format JSON

Azure Data Explorer prend en charge deux formats de fichier JSON :

json: JSON séparé par une ligne. Chaque ligne des données d’entrée contient exactement un enregistrement JSON. Ce format prend en charge l’analyse des commentaires et des propriétés entre guillemets uniques. Pour plus d’informations, consultez JSON Lines.
multijson: JSON multiligne. L’analyseur ignore les séparateurs de ligne et lit un enregistrement de la position précédente jusqu’à la fin d’un JSON valide.

Notes

Lors de l’ingestion à l’aide de l’Assistant Ingestion, le format par défaut est multijson. Le format peut gérer les enregistrements JSON multilignes et les tableaux d’enregistrements JSON. Lorsqu’une erreur d’analyse est rencontrée, le fichier entier est ignoré. Pour ignorer les enregistrements JSON non valides, sélectionnez l’option « Ignorer les erreurs de format de données », ce qui basculera le format vers json (lignes JSON).

Si vous utilisez le format de ligne JSON (json), les lignes qui ne représentent pas d’enregistrements JSON valides sont ignorées pendant l’analyse.

Ingérer et mapper des données au format JSON

L’ingestion de données au format JSON vous oblige à spécifier le format à l’aide de la propriété d’ingestion. L’ingestion de données JSON requiert un mappage, lequel mappe une entrée de source JSON à sa colonne cible. Lors d’une ingestion de données, utilisez la propriété IngestionMapping avec sa propriété d’ingestion ingestionMappingReference (pour un mappage prédéfini) ou sa propriété IngestionMappings. Cet article utilise la propriété d’ingestion ingestionMappingReference, qui est prédéfinie sur la table utilisée pour l’ingestion. Dans les exemples ci-dessous, nous allons commencer par ingérer des enregistrements JSON en tant que données brutes dans une table à une seule colonne. Nous utiliserons ensuite le mappage pour ingérer chaque propriété dans sa colonne mappée.

Exemple JSON simple

L’exemple suivant est un JSON simple, avec une structure plate. Les données comportent des informations sur la température et l’humidité, collectées par plusieurs appareils. Chaque enregistrement est marqué d’un ID et d’un timestamp.

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

Ingérer des enregistrements JSON bruts

Dans cet exemple, vous ingérez des enregistrements JSON en tant que données brutes dans une table à une seule colonne. La manipulation des données, l’utilisation de requêtes et la stratégie de mise à jour sont effectuées une fois que les données sont ingérées.

Utilisez Langage de requête Kusto pour ingérer des données dans un format JSON brut.

Connectez-vous à https://dataexplorer.azure.com.
Sélectionnez Ajouter un cluster.
Dans la boîte de dialogue Ajouter un cluster, entrez l’URL de votre cluster sous la forme https://<ClusterName>.<Region>.kusto.windows.net/, puis sélectionnez Ajouter.
Collez la commande suivante, puis sélectionnez Exécuter pour créer la table.
```
.create table RawEvents (Event: dynamic)
```
Cette requête crée la table munie d’une seule colonne Event d’un type de donnéesdynamique.
Créez le mappage JSON.
```
.create table RawEvents ingestion json mapping 'RawEventMapping' '[{"column":"Event","Properties":{"path":"$"}}]'
```
Cette commande crée un mappage et mappe le chemin d’accès racine JSON $ à la colonne Event.

Ingérez des données dans la table RawEvents.

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

Utilisez C# pour ingérer des données dans un format JSON brut.

Créez la table RawEvents.

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);

Créez le mappage 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);

Cette commande crée un mappage et mappe le chemin d’accès racine JSON $ à la colonne Event.

Ingérez des données dans la table RawEvents.

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);

Notes

Les données sont agrégées conformément à la stratégie de traitement par lot, ce qui entraîne une latence de quelques minutes.

Utilisez Python pour ingérer des données dans un format JSON brut.

Créez la table RawEvents.

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])

Créez le mappage 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])

Ingérez des données dans la table RawEvents.

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)

Notes

Les données sont agrégées conformément à la stratégie de traitement par lot, ce qui entraîne une latence de quelques minutes.

Ingérer des enregistrements JSON mappés

Dans cet exemple, vous ingérez des données d’enregistrements JSON. Chaque propriété JSON est mappée à une colonne unique de la table.

Créez une nouvelle table, avec un schéma similaire aux données d’entrée JSON. Nous utiliserons cette table pour tous les exemples et commandes d’ingestion suivants.
```
.create table Events (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)
```

Créez le mappage 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"}}]'

Dans ce mappage, comme défini par le schéma de la table, les entrées timestamp sont ingérées dans la colonne Time en tant que types de données datetime.

Ingérez des données dans la table Events.
```
.ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"FlatEventMapping"}'
```
Le fichier « simple.json » comporte quelques enregistrements JSON séparés par des lignes. Le format est json, et le mappage utilisé dans la commande d’ingestion est le FlatEventMapping que vous avez créé.

Créez une nouvelle table, avec un schéma similaire aux données d’entrée JSON. Nous utiliserons cette table pour tous les exemples et commandes d’ingestion suivants.

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);

Créez le mappage 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);

Dans ce mappage, comme défini par le schéma de la table, les entrées timestamp sont ingérées dans la colonne Time en tant que types de données datetime.

Ingérez des données dans la table Events.

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);

Le fichier « simple.json » comporte quelques enregistrements JSON séparés par des lignes. Le format est json, et le mappage utilisé dans la commande d’ingestion est le FlatEventMapping que vous avez créé.

Créez une nouvelle table, avec un schéma similaire aux données d’entrée JSON. Nous utiliserons cette table pour tous les exemples et commandes d’ingestion suivants.

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])

Créez le mappage 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])

Ingérez des données dans la table Events.

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)

Ingérer des enregistrements JSON multilignes

Dans cet exemple, vous ingérez des enregistrements JSON multilignes. Chaque propriété JSON est mappée à une colonne unique de la table. Le fichier « multilined.json » comporte quelques enregistrements JSON mis en retrait. Le format multijson indique la lecture des enregistrements par la structure JSON.

Ingérez des données dans la table Events.

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

Ingérez des données dans la table Events.

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);

Ingérez des données dans la table Events.

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)

Ingérer des enregistrements JSON contenant des tableaux

Les données de type tableau sont des collections ordonnées de valeurs. L’ingestion d’un tableau JSON est effectuée par une stratégie de mise à jour. Le JSON est ingéré tel quel dans une table intermédiaire. Une stratégie de mise à jour exécute une fonction prédéfinie sur la table RawEvents, en ingérant de nouveau les résultats dans la table cible. Nous allons ingérer les données avec la structure suivante :

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

Créez une fonction update policy qui développe la collection de records pour que chaque valeur de la collection reçoive une ligne distincte, à l’aide de l’opérateur mv-expand. Nous utiliserons la table RawEvents en tant que table source et Events comme table cible.

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

Le schéma reçu par la fonction doit correspondre au schéma de la table cible. Utilisez l’opérateur getschema pour examiner le schéma.
```
EventRecordsExpand() | getschema
```
Ajoutez la stratégie de mise à jour à la table cible. Cette stratégie exécute automatiquement la requête sur toutes les nouvelles données ingérées dans la table intermédiaire RawEvents et ingère ses résultats dans la table Events. Définissez une stratégie de rétention zéro pour éviter la persistance de la table intermédiaire.
```
.alter table Events policy update @'[{"Source": "RawEvents", "Query": "EventRecordsExpand()", "IsEnabled": "True"}]'
```

Ingérez des données dans la table RawEvents.

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

Examinez les données dans la table Events.
```
Events
```

Créez une fonction de mise à jour qui développe la collection de records pour que chaque valeur de la collection reçoive une ligne distincte, à l’aide de l’opérateur mv-expand. Nous utiliserons la table RawEvents en tant que table source et Events comme table cible.

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);

Notes

Le schéma reçu par la fonction doit correspondre au schéma de la table cible.

Ajoutez la stratégie de mise à jour à la table cible. Cette stratégie exécute automatiquement la requête sur toutes les nouvelles données ingérées dans la table intermédiaire RawEvents et ingère ses résultats dans la table Events. Définissez une stratégie de rétention zéro pour éviter la persistance de la table intermédiaire.
```
command = ".alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]";
await kustoClient.ExecuteControlCommandAsync(command);
```

Ingérez des données dans la table RawEvents.

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);

Examinez les données dans la table Events.

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])

Notes

Le schéma reçu par la fonction doit correspondre au schéma de la table cible.

Ajoutez la stratégie de mise à jour à la table cible. Cette stratégie exécute automatiquement la requête sur toutes les nouvelles données ingérées dans la table intermédiaire RawEvents et ingère ses résultats dans la table Events. Définissez une stratégie de rétention zéro pour éviter la persistance de la table intermédiaire.
```
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])
```

Ingérez des données dans la table RawEvents.

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)

Examinez les données dans la table Events.

Partager via

Ingérer des exemples de données au format JSON dans Azure Data Explorer

Prérequis

Le format JSON

Ingérer et mapper des données au format JSON

Exemple JSON simple

Ingérer des enregistrements JSON bruts

Ingérer des enregistrements JSON mappés

Ingérer des enregistrements JSON multilignes

Ingérer des enregistrements JSON contenant des tableaux

Commentaires

Ressources supplémentaires

Partager via

Ingérer des exemples de données au format JSON dans Azure Data Explorer

Prérequis

Le format JSON

Ingérer et mapper des données au format JSON

Exemple JSON simple

Ingérer des enregistrements JSON bruts

Ingérer des enregistrements JSON mappés

Ingérer des enregistrements JSON multilignes

Ingérer des enregistrements JSON contenant des tableaux

Contenu connexe

Commentaires

Ressources supplémentaires