你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn。
使用 Azure 数据资源管理器 Node 库引入数据
Azure 数据资源管理器是一项快速且高度可缩放的数据探索服务,适用于日志和遥测数据。 Azure 数据资源管理器为 Node 提供了两个客户端库:引入库和数据库。 可以使用这些库在群集中引入(加载)数据并从代码中查询数据。 本文首先在测试群集中创建一个表和数据映射。 然后将引入排列到群集并验证结果。
如果还没有 Azure 订阅,可以在开始前创建一个免费 Azure 帐户。
先决条件
安装数据和引入库
安装 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。 若要查找租户 ID,请按查找 Microsoft 365 租户 ID 中的说明操作。
为 authorityId
、kustoUri
、kustoIngestUri
和 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";
设置源文件信息
导入其他类并设置数据源文件的常数。 此示例使用 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.windows.net/${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.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