使用 Fluent Bit 将数据引入到 Azure 数据资源管理器

2025-06-13

Fluent Bit 是一种开源代理，用于从各种源收集日志、指标和跟踪。它让你可以在将事件数据发送到存储之前对事件数据进行筛选、修改和聚合。本文指导你完成使用 Fluent Bit 将数据发送到 KQL 数据库的过程。

本文介绍了如何使用 Fluent Bit 引入数据。

有关数据连接器的完整列表，请参阅数据连接器概述。

先决条件

Fluent Bit。
Azure 数据资源管理器群集和数据库。创建群集和数据库。
查询环境。有关详细信息，请参阅查询集成概览。
Kusto 群集 URI，用于表示 Ingestion_endpoint 值，格式为 https://ingest-<cluster>.<region>.kusto.chinacloudapi.cn。有关详细信息，请参阅添加群集连接。

创建 Microsoft Entra 服务主体

Microsoft Entra 服务主体可以通过 Azure 门户或以编程方式进行创建，如下例所示。

此服务主体是连接器用来将数据写入 Kusto 中的表的标识。授予此服务主体访问 Kusto 资源所需的权限。

通过 Azure CLI 登录到你的 Azure 订阅。然后在浏览器中进行身份验证。
```
az login
```
选择要托管主体的订阅。当你有多个订阅时，此步骤是必需的。
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```

创建服务主体。在此示例中，服务主体名为 my-service-principal。

az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}

从返回的 JSON 数据中复制 appId、password、tenant 供将来使用。

{
  "appId": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "tenant": "00001111-aaaa-2222-bbbb-3333cccc4444"
}

现已创建了 Microsoft Entra 应用程序和服务主体。

创建目标表

Fluent Bit 以 JSON 格式将日志随以下三个属性转发：log (dynamic)、tag (string) 和 timestamp (datetime)。

可以创建一个表，其中包含上述每个属性的列。或者，如果你有结构化日志，则可以创建一个表，其中包含映射到自定义列的日志属性。若要了解详细信息，请选择相关选项卡。

默认架构
自定义架构

若要为来自 Fluent Bit 的传入日志创建表，请执行以下操作：

浏览至查询环境。
选择要在其中创建表的数据库。
运行以下 .create table 命令：
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
传入的 JSON 属性会自动映射到正确的列中。

若要为来自 Fluent Bit 的传入结构化日志创建表，请执行以下操作：

浏览至查询环境。
选择要在其中创建表的数据库。
运行 .create table 命令。例如，如果日志包含名为 myString、myInteger 和 myDynamic 的三个字段，则可以创建具有以下架构的表：
```
.create table FluentBitLogs (myString:string, myInteger:int, myDynamic: dynamic, timestamp:datetime)
```

创建 JSON 映射，以将日志属性映射到相应的列。以下命令基于上一步中的示例创建映射：

.create-or-alter table FluentBitLogs ingestion json mapping "LogMapping" 
    ```[
    {"column" : "myString", "datatype" : "string", "Properties":{"Path":"$.log.myString"}},
    {"column" : "myInteger", "datatype" : "int", "Properties":{"Path":"$.log.myInteger"}}, 
    {"column" : "myDynamic", "datatype" : "dynamic", "Properties":{"Path":"$.log.myInteger"}}, 
    {"column" : "timestamp", "datatype" : "datetime", "Properties":{"Path":"$.timestamp"}} 
    ]```

向服务主体授予权限

根据创建 Microsoft Entra 服务主体授予服务主体数据库引入者角色，并提供使用数据库的权限。有关详细信息，请参阅示例。将占位符 DatabaseName 替换为目标数据库的名称，将 ApplicationID 替换为创建 Microsoft Entra 服务主体时保存的 AppId 值。

.add database <DatabaseName> ingestors ('aadapp=<ApplicationID>;<TenantID>')

配置 Fluent Bit 以将日志发送到表

若要配置 Fluent Bit 以将日志发送到 Kusto 中的表格，请创建经典模式或 YAML 模式配置文件，其中包含以下输出属性：

字段	说明	需要	默认
名称	管道名称。		`azure_kusto`
租户ID	创建 Microsoft Entra 服务主体中的租户 ID。	✔️
客户编号	创建 Microsoft Entra 服务主体中的应用程序 ID。	✔️
客户端密钥	创建 Microsoft Entra 服务主体中的客户端密钥值（密码）。	✔️
managed_identity_client_id (托管身份客户端ID)	用于身份验证的托管标识的客户端 ID。	✔️
数据摄取端点	输入 Ingestion_Endpoint 所述的值。	✔️
数据库名称	包含日志表的数据库的名称。	✔️
表名	创建目标表中的表的名称。	✔️
摄入映射参考	创建目标表中的引入映射的名称。如果未创建引入映射，请从配置文件中删除该属性。
日志键	日志内容的密钥名称。例如：`log`。		`log`
include_tag_key	如果启用，则会将标记追加到输出中。		`On`
标签键	标记的密钥名称。如果 `include_tag_key` 为 false，则忽略。		`tag`
包含时间键	时间戳将追加到输出（如果已启用）。使用 `time_key` 属性。		`On`
时间键	日志记录中时间戳的密钥名称。如果 `include_time_key` 为 false，则忽略。		`timestamp`
数据接收端点连接超时	各种 Kusto 终结点的连接超时（以秒为单位）。		`60`
压缩已启用	如果已启用，则向 Kusto 发送压缩的 HTTP 有效负载 (gzip)。		`true`
资源摄取刷新间隔	Kusto 终结点的引入资源刷新间隔（以秒为单位）。
工作者	为此输出执行刷新操作的辅助角色数。		`0`
缓冲已启用	如果启用，在引入 Kusto 之前，将数据缓冲到磁盘中。		`Off`
缓冲路径	指定将存储缓冲数据的目录的位置（如果 `buffering_enabled` 为 `On`）。		`/tmp/fluent-bit/azure-kusto/`
upload_timeout	指定上传超时（如果 `buffering_enabled` 为 `On`）。即使文件大小低于限制，早于此日期的文件也会被处理。		`30m`
上传文件大小	指定要上传的文件的最大大小（如果 `buffering_enabled` 为 `On`）。		`200MB`
azure_kusto_buffer_key	Azure Kusto 缓冲区的密钥用于在`buffering_enabledOn`时标识插件实例。使用缓冲的多个 Azure Kusto 输出是必需的。		`key`
存储目录限制大小	存储缓冲数据的目录的最大大小（如果 `buffering_enabled` 为 `On`）。		`8GB`
缓冲文件提前删除	当 `buffering_enabled` 为 `On` 时，是否在成功创建 Blob 后提前删除缓冲文件。		`Off`
统一_标签	创建单个缓冲区文件（如果 `buffering_enabled` 为 `On`. ）		`On`
blob_uri_length	在引入到 Kusto 之前，请设置生成的 Blob URI 的长度。		`64`
计划程序最大重试次数	当`buffering_enabled`是`On`时，请设置使用调度程序进行传输的最大重试次数。		`3`
在最大上传错误时删除	当`buffering_enabled`为`On`时，是否在最大上传错误的情况下删除缓冲文件。		`Off`
IO_timeout	为上传配置 HTTP IO 超时。		`60s`

若要查看示例配置文件，请选择相关选项卡：

经典模式
YAML 模式

[SERVICE]
    Daemon Off
    Flush 1
    Log_Level trace
    HTTP_Server On
    HTTP_Listen 0.0.0.0
    HTTP_Port 2020
    Health_Check On

[INPUT]
    Name tail
    Path /var/log/containers/*.log
    Tag kube.*
    Mem_Buf_Limit 1MB
    Skip_Long_Lines On
    Refresh_Interval 10

[OUTPUT]
    [OUTPUT]
    Match *
    Name azure_kusto
    Tenant_Id <app_tenant_id>
    Client_Id <app_client_id>
    Client_Secret <app_secret>
    Ingestion_Endpoint https://ingest-<cluster>.<region>.kusto.chinacloudapi.cn
    Database_Name <database_name>
    Table_Name <table_name>
    Ingestion_Mapping_Reference <mapping_name>
    ingestion_endpoint_connect_timeout <ingestion_endpoint_connect_timeout>
    compression_enabled <compression_enabled>
    ingestion_resources_refresh_interval <ingestion_resources_refresh_interval>
    buffering_enabled On
    upload_timeout 2m
    upload_file_size 125M
    azure_kusto_buffer_key kusto1
    buffer_file_delete_early Off
    unify_tag On
    buffer_dir /var/log/
    store_dir_limit_size 16GB
    blob_uri_length 128
    scheduler_max_retries 3
    delete_on_max_upload_error Off
    io_timeout 60s

config:
  service: |
    [SERVICE]
        Daemon Off
        Flush 1
        Log_Level trace
        HTTP_Server On
        HTTP_Listen 0.0.0.0
        HTTP_Port 2020
        Health_Check On
        
  inputs: |
    [INPUT]
        Name tail
        Path /var/log/containers/*.log
        multiline.parser docker, cri
        Tag kube.*
        Mem_Buf_Limit 1MB
        Skip_Long_Lines On
        Refresh_Interval 10

  filters: |
    [FILTER]
        Name kubernetes
        Match kube.*
        Merge_Log On
        Merge_Log_key log_processed
        K8S-Logging.Parser On
        K8S-Logging.Exclude Off


  outputs: |
    [OUTPUT]
        [OUTPUT]
    Match *
    Name azure_kusto
    Tenant_Id <app_tenant_id>
    Client_Id <app_client_id>
    Client_Secret <app_secret>
    Ingestion_Endpoint https://ingest-<cluster>.<region>.kusto.chinacloudapi.cn
    Database_Name <database_name>
    Table_Name <table_name>
    Ingestion_Mapping_Reference <mapping_name>
    ingestion_endpoint_connect_timeout <ingestion_endpoint_connect_timeout>
    compression_enabled <compression_enabled>
    ingestion_resources_refresh_interval <ingestion_resources_refresh_interval>
    buffering_enabled On
    upload_timeout 2m
    upload_file_size 125M
    azure_kusto_buffer_key kusto1
    buffer_file_delete_early Off
    unify_tag On
    buffer_dir /var/log/
    store_dir_limit_size 16GB
    blob_uri_length 128
    scheduler_max_retries 3
    delete_on_max_upload_error Off
    io_timeout 60s

确认数据引入

数据到达表后，通过检查行计数来确认数据传输：
```
FluentBitLogs
| count
```
若要查看日志数据示例，请运行以下查询：
```
FluentBitLogs
| take 100
```

通过