Вставка образца данных в формате 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 как необработанные данные в таблицу с одним столбцом. Обработка данных с использованием запросов и политики обновления выполняется после того, как данные были загружены.

KQL

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

1. Выполните вход в https://dataexplorer.azure.com.

2. Выберите Добавить кластер.

3. В диалоговом окне Добавить кластер введите URL-адрес кластера в форму https://<ClusterName>.<Region>.kusto.windows.net/, а затем выберите Добавить.

4. Вставьте следующую команду и выберите Выполнить, чтобы создать таблицу.

   .create table RawEvents (Event: dynamic)
   

   Этот запрос создает таблицу с одним столбцом Event динамического типа данных.

5. Создание сопоставления JSON.

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

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

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

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

C#

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

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

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

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

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

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

2. Создание сопоставления 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. Прием данных в таблице 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 сопоставляется с одним столбцом в таблице.

KQL

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

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

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

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

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

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

C#

1. Создайте новую таблицу со схемой, аналогичной входным данным 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);
   

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

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

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

Python

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

   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. Создание сопоставления 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. Прием данных в таблице 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.

KQL

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

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

C#

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

Python

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

KQL

1. Создайте функцию 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"])
   }
   

2. Схема, полученная функцией, должна соответствовать схеме целевой таблицы. Используйте оператор getschema для просмотра схемы.

   EventRecordsExpand() | getschema
   

3. Добавьте политику обновления для целевой таблицы. Эта политика автоматически запускает запрос ко всем недавно полученным данным в промежуточной таблице RawEvents и вставляет результаты в таблицу Events. Определите политику нулевого хранения, чтобы избежать сохранения промежуточной таблицы.

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

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

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

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

   Events
   

C#

1. Создайте функцию обновления, которая расширяет коллекцию 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);
   

   Примечание.

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

2. Добавьте политику обновления для целевой таблицы. Эта политика автоматически выполнит запрос ко всем недавно полученным данным в промежуточной таблице RawEvents и загрузит его результаты в таблицу Events. Определите политику нулевого хранения, чтобы избежать сохранения промежуточной таблицы.

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

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

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

Python

1. Создайте функцию обновления, которая расширяет коллекцию 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])
   

   Примечание.

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

2. Добавьте политику обновления для целевой таблицы. Эта политика автоматически выполнит запрос ко всем недавно полученным данным в промежуточной таблице 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])
   

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

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

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

• Обзор приема данных
• Написание запросов