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

Прием данных с помощью библиотеки узлов Azure Data Explorer

Azure Data Explorer — это быстрая и высокомасштабируемая служба для изучения данных журналов и телеметрии. Azure Data Explorer предоставляет две клиентские библиотеки для Node: библиотеку загрузки данных и библиотеку данных. Эти библиотеки позволяют получать (загружать) данные в кластер и запрашивать данные из кода. В этой статье сначала вы создаете таблицу и сопоставление данных в тестовом кластере. Затем вы ставите в очередь обработку данных в кластер и проверяете результаты.

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

Предпосылки

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

Установка библиотек загрузки данных

Установка azure-kusto-ingest и azure-kusto-data

npm i azure-kusto-ingest@^3.3.2 azure-kusto-data@^3.3.2

Добавление операторов и констант импорта

Импорт классов из библиотек


const { Client: KustoClient, KustoConnectionStringBuilder } =  require('azure-kusto-data');
const {
    IngestClient: KustoIngestClient,
    IngestionProperties,
    IngestionDescriptors,
    DataFormat,
    IngestionMappingKind,
} =  require("azure-kusto-ingest");

Для проверки подлинности приложения Azure Data Explorer использует идентификатор клиента Microsoft Entra. Чтобы найти идентификатор клиента, выполните поиск идентификатора клиента Microsoft 365.

Задайте значения для authorityId, kustoUrikustoIngestUri и kustoDatabase перед выполнением этого кода.

const cluster = "MyCluster";
const region = "westus";
const authorityId = "microsoft.com";
const kustoUri = `https://${cluster}.${region}.kusto.windows.net`;
const kustoIngestUri = `https://ingest-${cluster}.${region}.kusto.windows.net`;
const kustoDatabase  = "Weather";

Создайте строку подключения. В этом примере используется проверка подлинности устройства для доступа к кластеру. Проверьте выходные данные консоли, чтобы завершить проверку подлинности. Вы также можете использовать сертификат приложения Microsoft Entra, ключ приложения, имя пользователя и пароль.

Вы создаете целевую таблицу и сопоставление на следующем шаге.

const kcsbIngest = KustoConnectionStringBuilder.withAadDeviceAuthentication(kustoIngestUri, authorityId);
const kcsbData = KustoConnectionStringBuilder.withAadDeviceAuthentication(kustoUri, authorityId);
const destTable = "StormEvents";
const destTableMapping = "StormEvents_CSV_Mapping";

Установка сведений о исходном файле

Импортируйте дополнительные классы и задайте константы для файла источника данных. В этом примере используется пример файла, размещенного в хранилище BLOB-объектов Azure. Пример набора данных StormEvents содержит данные, связанные с погодой, из Национальных центров по экологической информации.

const container = "samplefiles";
const account = "kustosamples";
const sas = "";  // If relevant add SAS token
const filePath = "StormEvents.csv";
const blobPath = `https://${account}.blob.core.windows.net/${container}/${filePath}${sas}`;

Создание таблицы в тестовом кластере

Создайте таблицу, которая соответствует схеме StormEvents.csv данных в файле. При выполнении этого кода возвращается следующее сообщение: для входа используйте веб-браузер, чтобы открыть страницу https://microsoft.com/devicelogin и ввести код XXXXXXXXXXX для проверки подлинности. Выполните действия для входа, а затем вернитесь к следующему блоку кода. Последующие блоки кода, которые делают подключение, потребуют повторного входа.

const kustoClient = new KustoClient(kcsbData);
const createTableCommand = `.create table ${destTable} (StartTime: datetime, EndTime: datetime, EpisodeId: int, EventId: int, State: string, EventType: string, InjuriesDirect: int, InjuriesIndirect: int, DeathsDirect: int, DeathsIndirect: int, DamageProperty: int, DamageCrops: int, Source: string, BeginLocation: string, EndLocation: string, BeginLat: real, BeginLon: real, EndLat: real, EndLon: real, EpisodeNarrative: string, EventNarrative: string, StormSummary: dynamic)`;

const createTableResults = await kustoClient.executeMgmt(kustoDatabase, createTableCommand);
console.log(createTableResults.primaryResults[0].toJSON().data);

Определение сопоставления приема

Сопоставляйте входящие csv-данные с именами столбцов и типами данных, используемыми при создании таблицы.

const createMappingCommand = `.create table ${destTable} ingestion csv mapping '${destTableMapping}' '[{"Name":"StartTime","datatype":"datetime","Ordinal":0}, {"Name":"EndTime","datatype":"datetime","Ordinal":1},{"Name":"EpisodeId","datatype":"int","Ordinal":2},{"Name":"EventId","datatype":"int","Ordinal":3},{"Name":"State","datatype":"string","Ordinal":4},{"Name":"EventType","datatype":"string","Ordinal":5},{"Name":"InjuriesDirect","datatype":"int","Ordinal":6},{"Name":"InjuriesIndirect","datatype":"int","Ordinal":7},{"Name":"DeathsDirect","datatype":"int","Ordinal":8},{"Name":"DeathsIndirect","datatype":"int","Ordinal":9},{"Name":"DamageProperty","datatype":"int","Ordinal":10},{"Name":"DamageCrops","datatype":"int","Ordinal":11},{"Name":"Source","datatype":"string","Ordinal":12},{"Name":"BeginLocation","datatype":"string","Ordinal":13},{"Name":"EndLocation","datatype":"string","Ordinal":14},{"Name":"BeginLat","datatype":"real","Ordinal":16},{"Name":"BeginLon","datatype":"real","Ordinal":17},{"Name":"EndLat","datatype":"real","Ordinal":18},{"Name":"EndLon","datatype":"real","Ordinal":19},{"Name":"EpisodeNarrative","datatype":"string","Ordinal":20},{"Name":"EventNarrative","datatype":"string","Ordinal":21},{"Name":"StormSummary","datatype":"dynamic","Ordinal":22}]'`;

const mappingCommandResults = await kustoClient.executeMgmt(kustoDatabase, createMappingCommand);
console.log(mappingCommandResults.primaryResults[0].toJSON().data);

Постановка сообщения в очередь для обработки

Поставьте в очередь сообщение для извлечения данных из облачного хранилища BLOB и переноса этих данных в Azure Data Explorer.

const defaultProps  = new IngestionProperties({
    database: kustoDatabase,
    table: destTable,
    format: DataFormat.CSV,
    ingestionMappingReference: destTableMapping,
    ingestionMappingKind: IngestionMappingKind.CSV,
    additionalProperties: {ignoreFirstRecord: true},
});

const ingestClient = new KustoIngestClient(kcsbIngest, defaultProps);
// All ingestion properties are documented here: https://learn.microsoft.com/azure/kusto/management/data-ingest#ingestion-properties

const blobDesc = new BlobDescriptor(blobPath, 10);
try {
	const ingestionResult = await ingestClient.ingestFromBlob(blobDesc, null);
} catch (err) {
	// Handle errors
}

Проверка того, что таблица содержит данные

Убедитесь, что данные были загружены в таблицу. Подождите пять–десять минут для приема в очереди, чтобы запланировать прием и загрузить данные в Azure Data Explorer. Затем выполните следующий код, чтобы получить количество записей в StormEvents таблице.

const query = `${destTable} | count`;

var tableResults = await kustoClient.execute(kustoDatabase, query);
console.log(tableResults.primaryResults[0].toJSON().data);

Выполнение запросов по устранению неполадок

Войдите в https://dataexplorer.azure.com и подключитесь к кластеру. Выполните следующую команду в базе данных, чтобы узнать, были ли ошибки загрузки за последние четыре часа. Замените имя базы данных перед выполнением.

.show ingestion failures
| where FailedOn > ago(4h) and Database == "<DatabaseName>"

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

.show operations
| where StartedOn > ago(4h) and Database == "<DatabaseName>" and Operation == "DataIngestPull"
| summarize arg_max(LastUpdatedOn, *) by OperationId

Очистите ресурсы

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

.drop table StormEvents

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

  • Last updated on