使用 Azure 数据资源管理器 Node 库引入数据

项目
12/22/2023

Azure 数据资源管理器是一项快速且高度可缩放的数据探索服务，适用于日志和遥测数据。 Azure 数据资源管理器为 Node 提供了两个客户端库：引入库和数据库。可以使用这些库在群集中引入（加载）数据并从代码中查询数据。本文首先在测试群集中创建一个表和数据映射。然后将引入排列到群集并验证结果。

如果还没有 Azure 订阅，可以在开始前创建一个 Azure 帐户。

先决条件

Microsoft 帐户或 Microsoft Entra 用户标识。无需 Azure 订阅。
Azure 数据资源管理器群集和数据库。创建群集和数据库。
已在开发计算机上安装了 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 数据资源管理器使用 Microsoft Entra 租户 ID 来对应用程序进行身份验证。

为 authorityId、kustoUri、kustoIngestUri 和 kustoDatabase 设置值，然后运行以下代码。

const cluster = "MyCluster";
const region = "chinaeast2";
const authorityId = "microsoft.com";
const kustoUri = `https://${cluster}.${region}.kusto.chinacloudapi.cn`;
const kustoIngestUri = `https://ingest-${cluster}.${region}.kusto.chinacloudapi.cn`;
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";

设置源文件信息

导入其他类并设置数据源文件的常数。此示例使用 Azure Blob 存储上托管的示例文件。 StormEvents 示例数据集包含美国国家环境信息中心的与天气相关的数据。

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

在测试群集上创建表

创建与 StormEvents.csv 文件中的数据架构匹配的表。运行此代码时，它会返回如下消息：若要登录，请使用 Web 浏览器打开页 https://microsoft.com/devicelogin ，然后输入代码 XXXXXXXXX 进行身份验证。按照步骤登录，然后返回运行下一个代码块。建立连接的后续代码块会要求你再次登录。

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 数据资源管理器。

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 数据资源管理器中引入和加载数据。然后运行以下代码，以获取 StormEvents 表中记录的计数。

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

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

运行故障排除查询

登录到 https://dataexplorer.azure.cn 并连接到群集。在数据库中运行以下命令以查看过去四个小时内是否存在任何失败引入。在运行之前替换数据库名称。

.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

编写查询