将数据从 Logstash 引入 Azure 数据资源管理器

项目
12/06/2023

重要

此连接器可用于 Microsoft Fabric 中的实时分析。使用本文中的说明时，请注意以下例外情况：

如果需要，请按照创建 KQL 数据库中的说明创建数据库。
如果需要，请按照创建空表中的说明创建表。
使用复制 URI 中的说明获取查询或引入 URI。
运行 KQL 查询集中的查询。

Logstash 是一个开源服务器端数据处理管道，可以同时从多个源引入数据、转换数据，然后将数据发送到你偏好的“储存”中。在本文中，你要将该数据发送到 Azure 数据资源管理器 - 用于日志和遥测数据的快速且高度可缩放的数据探索服务。首先你将在测试群集中创建一个表和数据映射，然后指示 Logstash 将数据发送到该表，并验证结果。

注意

此连接器目前仅支持 json 数据格式。

先决条件

Microsoft 帐户或 Microsoft Entra 用户标识。无需 Azure 订阅。
Azure 数据资源管理器群集和数据库。创建群集和数据库。
Logstash 版本 6+ 安装说明。

创建表

准备好群集和数据库之后，可以创建一个表。

在数据库查询窗口中运行以下命令，以创建表：
```
.create table logs (timestamp: datetime, message: string)
```
运行以下命令，确认已创建新表 logs 并且该表是空的：
```
logs
| count
```

创建映射

Azure 数据资源管理器使用映射将传入的数据转换为目标表架构。以下命令创建名为 basicmsg 的新映射。该映射根据 path 的指定从传入的 json 中提取属性，然后将这些属性输出到 column。

在查询窗口中运行以下命令：

.create table logs ingestion json mapping 'basicmsg' '[{"column":"timestamp","path":"$.@timestamp"},{"column":"message","path":"$.message"}]'

安装 Logstash 输出插件

Logstash 输出插件与 Azure 数据资源管理器通信，并将数据发送到服务。有关详细信息，请参阅 Logstash 插件。

在命令 shell 中，导航到 Logstash 根目录，然后运行以下命令以安装插件：

bin/logstash-plugin install logstash-output-kusto

配置 Logstash 以生成示例数据集

Logstash 可以生成用于测试端到端管道的示例事件。如果你正在使用 Logstash 并且有权访问自己的事件流，请跳到下一部分。

注意

如果使用自己的数据，请更改上一步骤中定义的表和映射对象。

编辑包含所需管道设置的新文本文件（使用 vi）：
```
vi test.conf
```

粘贴以下设置，告知 Logstash 生成 1000 个测试事件：

input {
    stdin { }
    generator {
        message => "Test Message 123"
        count => 1000
    }
}

此配置还包括 stdin 输入插件。使用该插件可以自行编写更多的消息（请务必使用 Enter 将消息提交到管道中）。

配置 Logstash 以将数据发送到 Azure 数据资源管理器

将以下设置粘贴到在上一步骤中使用的同一配置文件中。请将所有占位符替换为相关的设置值。有关详细信息，请参阅创建 Microsoft Entra 应用程序。

output {
    kusto {
            path => "/tmp/kusto/%{+YYYY-MM-dd-HH-mm-ss}.txt"
            ingest_url => "https://ingest-<cluster name>.kusto.chinacloudapi.cn/"
            app_id => "<application id>"
            app_key => "<application key/secret>"
            app_tenant => "<tenant id>"
            database => "<database name>"
            table => "<target table>" # logs as defined above
            json_mapping => "<mapping name>" # basicmsg as defined above
    }
}

参数名称	说明
路径	Logstash 插件会将事件写入临时文件，然后将其发送到 Azure 数据资源管理器。此参数包含要将文件写入到的路径，以及一个用于轮转文件的时间表达式，该表达式可触发上传到 Azure 数据资源管理器服务的操作。
ingest_url	用于进行引入相关通信的 Kusto 终结点。
app_id、app_key 和 app_tenant	连接到 Azure 数据资源管理器所需的凭据。请务必使用具有引入特权的应用程序。
database	要将事件放入到的数据库的名称。
table	要将事件放入到的目标表的名称。
json_mapping	mapping（映射）用于将传入事件的 json 字符串映射为正确的行格式（定义哪些属性要进入哪个列）。

运行 Logstash

现在，可以运行 Logstash 并测试设置。

在命令 shell 中，导航到 Logstash 根目录，然后运行以下命令：
```
bin/logstash -f test.conf
```
屏幕上应会输出信息，然后输出示例配置生成的 1000 条消息。此时还可以手动输入更多消息。
几分钟后，请运行以下数据资源管理器查询来查看所定义的表中的消息：
```
logs
| order by timestamp desc
```
按 Ctrl+C 退出 Logstash

清理资源

在数据库中运行以下命令以清理 logs 表：

.drop table logs

编写查询