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

Статья
10/29/2023

В этой статье показано, как загрузить данные в формате 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.

Поделиться через

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

Предварительные требования

Формат JSON

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

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

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

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

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

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

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

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

Поделиться через

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

Предварительные требования

Формат JSON

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

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

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

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

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

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

См. также

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

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