Вставка образца данных в формате JSON в Azure Data Explorer

В этой статье показано, как загрузить данные в формате JSON в базу данных Azure Data Explorer. Вы начинаете с простых примеров необработанного и сопоставленного JSON, продолжайте использовать многострочный json, а затем решаете более сложные схемы JSON, содержащие массивы и словари. В примерах подробно описан процесс приема форматированных данных JSON с помощью язык запросов Kusto (KQL), C#или Python.

Примечание.

Мы не рекомендуем использовать .ingest команды управления в рабочих сценариях. Вместо этого используйте соединитель данных или программным способом прием данных с помощью одной из клиентских библиотек Kusto.

Необходимые компоненты

Учетная запись Майкрософт или удостоверение пользователя Microsoft Entra. Подписка Azure не обязательна.
Кластер и база данных Azure Data Explorer. Создайте кластер и базу данных.

Формат JSON

Azure Data Explorer поддерживает два формата файлов JSON.

json: разделенный строкой JSON. Каждая строка входных данных содержит ровно одну запись JSON. Этот формат поддерживает синтаксический анализ комментариев и одноцитированных свойств. Для получения дополнительной информации см. Строки JSON.
multijson: многострочный JSON. Парсер игнорирует разделители строк и считывает запись с предыдущей позиции до конца допустимого JSON.

Примечание.

При приеме с помощью интерфейса получения данных используется multijsonформат по умолчанию. Формат может обрабатывать многостроные записи JSON и массивы записей JSON. При обнаружении ошибки синтаксического анализа весь файл удаляется. Чтобы игнорировать недопустимые записи JSON, выберите параметр "Игнорировать ошибки формата данных"., который переключает формат json на (строки JSON).

Если вы используете формат строки JSON (json), строки, которые не представляют допустимые записи JSON, пропускаются во время синтаксического анализа.

Получение и сопоставление данных в формате JSON

При приеме данных в формате JSON необходимо указать формат с помощью свойства приема данных. Для приема данных JSON требуется сопоставление, которое сопоставляет исходную запись JSON с ее целевым столбцом. При приеме данных используйте IngestionMapping свойство с его ingestionMappingReference свойством приема (для предопределенного сопоставления) приема или его IngestionMappings свойства. В этой статье используется ingestionMappingReference свойство приема, которое предопределено в таблице, используемой для приема. В следующих примерах мы начинаем прием записей JSON в виде необработанных данных в одну таблицу столбцов. Затем мы используем сопоставление для приема каждого свойства в сопоставленный столбец.

Простой пример JSON

Следующий пример представляет собой простой JSON с плоской структурой. Данные содержат информацию о температуре и влажности, собираемую несколькими устройствами. Каждая запись помечается идентификатором и отметкой времени.

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

Получение необработанных записей JSON

В этом примере вы загружаете записи JSON как необработанные данные в таблицу с одним столбцом. Обработка данных с использованием запросов и политики обновления выполняется после того, как данные были загружены.

Используйте язык запросов Kusto для приема данных в необработанном формате JSON.

Выполните вход в https://dataexplorer.azure.com.
Выберите Добавить кластер.
В диалоговом окне Добавить кластер введите URL-адрес кластера в форму https://<ClusterName>.<Region>.kusto.windows.net/, а затем выберите Добавить.
Вставьте следующую команду и выберите Выполнить, чтобы создать таблицу.
```
.create table RawEvents (Event: dynamic)
```
Этот запрос создает таблицу с одним столбцом Eventдинамического типа данных.
Создание сопоставления JSON.
```
.create table RawEvents ingestion json mapping 'RawEventMapping' '[{"column":"Event","Properties":{"path":"$"}}]'
```
Эта команда создает сопоставление и сопоставляет корневой путь JSON $ столбцу Event.

Прием данных в таблице RawEvents.

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

Используйте C# для приема данных в необработанном формате JSON.

Создание 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);

Создание сопоставления 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);

Эта команда создает сопоставление и сопоставляет корневой путь JSON $ столбцу Event.

Прием данных в таблице 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);

Примечание.

Статистические данные вычисляются в соответствии с политикой пакетной обработки, что приводит к задержке в несколько минут.

Используйте Python для приема данных в необработанном формате JSON.

Создание 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])

Создание сопоставления 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])

Прием данных в таблице 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)

Примечание.

Прием сопоставленных записей JSON

В этом примере вы загружаете данные записей JSON. Каждое свойство JSON сопоставляется с одним столбцом в таблице.

Создайте новую таблицу со схемой, аналогичной входным данным JSON. Мы используем эту таблицу для всех приведенных ниже примеров и команд приема.
```
.create table Events (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)
```

Создание сопоставления 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"}}]'

В этом сопоставлении, как определено схемой таблицы, timestamp записи отправляются в столбец Time как datetime типы данных.

Прием данных в таблице Events.
```
.ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"FlatEventMapping"}'
```
В файле simple.json есть несколько записей JSON, разделенных строками. Формат — json, а отображение, используемое в команде приема данных, — это созданное вами FlatEventMapping.

Создайте новую таблицу со схемой, аналогичной входным данным JSON. Мы используем эту таблицу для всех приведенных ниже примеров и команд приема.

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

Создание сопоставления 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);

Прием данных в таблице 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);

В файле simple.json есть несколько записей JSON, разделенных строками. Формат — json, а отображение, используемое в команде приема данных, — это созданное вами FlatEventMapping.

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

Создание сопоставления 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])

Прием данных в таблице 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)

Файл simple.json содержит несколько записей JSON, разделенных строками. Формат — json, а отображение, используемое в команде приема данных, — это созданное вами FlatEventMapping.

Получение многострочных записей JSON

В этом примере вы загружаете многострочные записи JSON. Каждое свойство JSON сопоставляется с одним столбцом в таблице. В файле Multiined.json есть несколько записей JSON с отступом. multijson Формат указывает на чтение записей по структуре JSON.

Прием данных в таблице Events.

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

Прием данных в таблице 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);

Прием данных в таблице 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)

Принимать записи JSON, содержащие массивы

Тип данных "массив" представляет собой упорядоченную коллекцию значений. Получение массива JSON выполняется политикой обновления. JSON загружается как есть в промежуточную таблицу. Политика обновления запускает предопределенную функцию в RawEvents таблице, повторно возвращая результаты в целевую таблицу. Мы прием данных со следующей структурой:

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

Создайте функцию update policy, которая расширяет коллекцию records, чтобы каждое значение в коллекции получало отдельную строку, используя оператор mv-expand. Мы используем таблицу RawEvents в качестве исходной таблицы и Events в качестве целевой таблицы.
```
.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"])
}
```
Схема, полученная функцией, должна соответствовать схеме целевой таблицы. Используйте оператор getschema для просмотра схемы.
```
EventRecordsExpand() | getschema
```
Добавьте политику обновления для целевой таблицы. Эта политика автоматически запускает запрос на все недавно полученные данные в промежуточной RawEvents таблице и возвращает результаты в таблицу Events . Определите политику нулевого хранения, чтобы избежать сохранения промежуточной таблицы.
```
.alter table Events policy update @'[{"Source": "RawEvents", "Query": "EventRecordsExpand()", "IsEnabled": "True"}]'
```

Прием данных в таблице RawEvents.

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

Просмотрите данные в таблице Events.
```
Events
```

Создайте функцию обновления, которая расширяет коллекцию records, чтобы каждое значение в коллекции получало отдельную строку, используя оператор mv-expand. Мы используем таблицу RawEvents в качестве исходной таблицы и Events в качестве целевой таблицы.
```
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);
```
Примечание.

Схема, полученная функцией, должна соответствовать схеме целевой таблицы.
Добавьте политику обновления для целевой таблицы. Эта политика автоматически запускает запрос на все недавно полученные данные в промежуточной RawEvents таблице и возвращает результаты в таблицу Events . Определите политику нулевого хранения, чтобы избежать сохранения промежуточной таблицы.
```
command = ".alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]";
await kustoClient.ExecuteControlCommandAsync(command);
```

Прием данных в таблице 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);

Просмотрите данные в таблице Events.

Создайте функцию обновления, которая расширяет коллекцию records, чтобы каждое значение в коллекции получало отдельную строку, используя оператор mv-expand. Мы используем таблицу RawEvents в качестве исходной таблицы и 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])
```
Примечание.

Схема, полученная функцией, должна соответствовать схеме целевой таблицы.
Добавьте политику обновления для целевой таблицы. Эта политика автоматически запускает запрос на все недавно полученные данные в промежуточной RawEvents таблице и возвращает результаты в таблицу Events . Определите политику нулевого хранения, чтобы избежать сохранения промежуточной таблицы.
```
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])
```

Прием данных в таблице 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)

Просмотрите данные в таблице Events.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2025-09-02

Вставка образца данных в формате JSON в Azure Data Explorer

Необходимые компоненты

Формат JSON

Получение и сопоставление данных в формате JSON

Простой пример JSON

Получение необработанных записей JSON

Прием сопоставленных записей JSON

Получение многострочных записей JSON

Принимать записи JSON, содержащие массивы

Связанный контент

Обратная связь

Дополнительные ресурсы