使用 Fluent Bit 将数据引入到 Azure Data Explorer

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

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

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

Prerequisites

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

创建 Microsoft Entra 服务主体

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

连接器使用此服务主体作为标识将数据写入 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 服务主体的 database ingestor 角色中授予服务主体权限，以便能够使用数据库。有关详细信息，请参阅示例。将占位符 DatabaseName 替换为目标数据库的名称，将 ApplicationID 替换为创建 Microsoft Entra 服务主体时保存的 AppId 值。

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

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

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

Field	Description	Required	Default
Name	管道名称。		`azure_kusto`
tenant_id	在创建 Microsoft Entra 服务主体中获取的租户 ID。	✔️
client_id	创建 Microsoft Entra 服务主体中的应用程序 ID。	✔️
client_secret	创建 Microsoft Entra 服务主体中的客户端密钥值（密码）。	✔️
managed_identity_client_id	用于身份验证的托管标识的客户端 ID。	✔️
ingestion_endpoint	输入 Ingestion_Endpoint 所述的值。	✔️
database_name	包含日志表的数据库的名称。	✔️
table_name	创建目标表中的表的名称。	✔️
ingestion_mapping_reference	创建目标表中的引入映射的名称。如果未创建引入映射，请从配置文件中删除该属性。
log_key	日志内容的密钥名称。例如：`log`。		`log`
include_tag_key	如果启用，则会将标记追加到输出中。		`On`
tag_key	标记的密钥名称。如果 `include_tag_key` 为 false，则忽略。		`tag`
include_time_key	时间戳将追加到输出（如果已启用）。使用 `time_key` 属性。		`On`
time_key	日志记录中时间戳的密钥名称。如果 `include_time_key` 为 false，则忽略。		`timestamp`
ingestion_endpoint_connect_timeout	各种 Kusto 终结点的连接超时（以秒为单位）。		`60`
compression_enabled	如果已启用，则向 Kusto 发送压缩的 HTTP 有效负载 (gzip)。		`true`
ingestion_resources_refresh_interval	Kusto 终结点的引入资源刷新间隔（以秒为单位）。
workers	为此输出执行刷新操作的工作线程数。		`0`
buffering_enabled	如果启用，在引入 Kusto 之前，将数据缓冲到磁盘中。		`Off`
buffer_path	指定将存储缓冲数据的目录的位置（如果 `buffering_enabled` 为 `On`）。		`/tmp/fluent-bit/azure-kusto/`
upload_timeout	指定上传超时（如果 `buffering_enabled` 为 `On`）。即使文件大小低于限制，早于此日期的文件也会被处理。		`30m`
upload_file_size	指定要上传的文件的最大大小（如果 `buffering_enabled` 为 `On`）。		`200MB`
azure_kusto_buffer_key	Azure Kusto 缓冲区密钥，用于在 `buffering_enabled` 为 `On`时标识插件实例。实现多个 Azure Kusto 输出需要进行缓冲。		`key`
store_dir_limit_size	存储缓冲数据的目录的最大大小（如果 `buffering_enabled` 为 `On`）。		`8GB`
buffer_file_delete_early	当 `buffering_enabled` 为 `On` 时，是否在成功创建 Blob 后提前删除缓冲文件。		`Off`
unify_tag	创建单个缓冲区文件（如果 `buffering_enabled` 为 `On`. ）		`On`
blob_uri_length	在引入到 Kusto 之前，请设置生成的 Blob URI 的长度。		`64`
scheduler_max_retries	当`buffering_enabled`是`On`时，请设置使用调度程序进行传输的最大重试次数。		`3`
delete_on_max_upload_error	当`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
```

上次更新时间为 2026-03-10