使用 Fluent Bit 將數據內嵌至 Azure 數據總管

Fluent Bit 是開放原始碼代理程式，其會收集來自各種來源的記錄、計量和追蹤。它可讓您先篩選、修改和彙總事件資料，再將其傳送至儲存體。本文會引導您完成使用 Fluent Bit 將資料傳送至 KQL 資料庫的程序。

本文說明如何使用 Fluent Bit 內嵌數據。

如需資料連接器的完整清單，請參閱資料連接器概觀。

必要條件

Fluent Bit。
Azure 資料總管叢集和資料庫。建立叢集和資料庫。
查詢環境。如需詳細資訊，請參閱查詢整合概觀。
您的 Kusto 叢集 URI 用於 Ingestion_endpoint 值的格式：。如需詳細資訊，請參閱新增叢集連線。

建立 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`
租戶識別碼	從建立 Microsoft Entra 服務主體的租戶 ID。	✔️
client_id (客戶識別碼)	來自建立 Microsoft Entra 服務主體的應用程式識別碼。	✔️
用戶端密鑰	建立 Microsoft Entra 服務主體的用戶端秘密密鑰值（password）。	✔️
托管身份客戶端ID (managed_identity_client_id)	要用於驗證之受控識別的用戶端標識碼。	✔️
資料輸入端點	請依據 Ingestion_Endpoint的說明輸入值。	✔️
資料庫名稱	包含您的記錄資料表的資料庫名稱。	✔️
表格名稱	建立目標數據表的數據表名稱。	✔️
資料導入映射參考	從建立目標數據表的擷取對應名稱。如果您未建立擷取對應，請從設定檔中移除該屬性。
log_key	日誌內容的鍵名。例如，`log`。		`log`
包含標籤鍵	如果啟用，標記會附加至輸出。		`On`
標籤鍵	標記的索引鍵名稱。如果 `include_tag_key` 為 false，則會忽略。		`tag`
包含時間鍵	如果已啟用，時間戳會附加至輸出。使用 `time_key` 屬性。		`On`
時間鍵	記錄檔記錄中時間戳的索引鍵名稱。如果 `include_time_key` false，則會忽略。		`timestamp`
資料接入端點連接超時	不同 Kusto 端點的連線逾時，以秒為單位。		`60`
壓縮_已啟用	如果已啟用，會將壓縮的HTTP承載(gzip)傳送至Kusto。		`true`
資料攝取資源更新間隔	Kusto 端點的資料擷取資源重新整理間隔（以秒為單位）。
工人	為了執行排清作業，此輸出需要的工人數量為。		`0`
緩衝啟用	如果啟用, 便先將數據緩衝到磁碟，然後再導入 Kusto。		`Off`
緩衝路徑	指定如果 `buffering_enabled` 為 `On`，則會存放緩衝資料的目錄位置。		`/tmp/fluent-bit/azure-kusto/`
上傳超時	如果`buffering_enabled`為`On`，則指定上傳的逾時。即使檔案大小低於限制，比這個還舊的檔案也將被處理。		`30m`
上傳檔案大小	指定如果 `buffering_enabled` 為 `On`，則要上傳的檔案大小上限。		`200MB`
Azure 叢集緩衝區鍵	Azure Kusto 緩衝區密鑰，用於當`buffering_enabled`是`On`時識別外掛程式實例。若要使用多個 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 輸入/輸出逾時設定。		`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.windows.net
    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.windows.net
    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
```

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-06-26