Ingesta de datos con Fluent Bit en Azure Data Explorer

Fluent Bit es un agente de código abierto que recopila registros, métricas y seguimientos de varios orígenes. Permite filtrar, modificar y agregar datos de eventos antes de enviarlos al almacenamiento. Este artículo le guía por el proceso de uso de Fluent Bit para enviar datos a la base de datos KQL.

En este artículo se muestra cómo ingerir datos con Fluent Bit.

Para obtener una lista completa de los conectores de datos, consulte Introducción a los conectores de datos.

Requisitos previos

Fluent Bit.
Un clúster y la base de datos de Azure Data Explorer. Cree un clúster y una base de datos.
Un entorno de consulta. Para más información, vea la Introducción a las integraciones de consulta.
El URI del clúster de Kusto para el valor Ingestion_endpoint en el formato https://ingest-<clúster>.<region>.kusto.windows.net. Para más información, consulte Agregar una conexión de clúster.

Creación de una entidad de servicio de Microsoft Entra

La entidad de servicio de Microsoft Entra se puede crear mediante Azure Portal o mediante programación, como en el ejemplo siguiente.

Esta entidad de servicio es la identidad que usa el conector para escribir datos en la tabla en Kusto. Se te conceden permisos para que esta entidad de servicio acceda a los recursos de Kusto.

Inicie sesión en su suscripción de Azure a través de la CLI de Azure. A continuación, realice la autenticación en el explorador.
```
az login
```
Elija la suscripción para hospedar la entidad de servicio. Este paso es necesario si tiene varias suscripciones.
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```

Cree la entidad de servicio. En este ejemplo, la entidad de servicio se llama my-service-principal.

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

En los datos JSON devueltos, copie los valores appId, password y tenant para usarlos posteriormente.

{
  "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"
}

Ha creado una aplicación de Microsoft Entra y una entidad de servicio.

Creación de una tabla de destino

Fluent Bit reenvía los registros en formato JSON con tres propiedades: log (dynamic), tag (string) y timestamp (datetime).

Puede crear una tabla con columnas para cada una de estas propiedades. Como alternativa, si tiene registros estructurados, puede crear una tabla con propiedades de registro asignadas a columnas personalizadas. Para más información, seleccione la pestaña correspondiente.

Esquema predeterminado
Esquema personalizado

Para crear una tabla para los registros entrantes de Fluent Bit:

Vaya al entorno de consulta.
Seleccione la base de datos donde desea crear la tabla.
Ejecute el comando .create table siguiente:
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
Las propiedades JSON entrantes se asignan automáticamente a la columna correcta.

Para crear una tabla para los registros estructurados entrantes de Fluent Bit:

Vaya al entorno de consulta.
Seleccione la base de datos donde desea crear la tabla.
Ejecute el comando .create table. Por ejemplo, si los registros contienen tres campos denominados myString, myInteger y myDynamic, puede crear una tabla con el esquema siguiente:
```
.create table FluentBitLogs (myString:string, myInteger:int, myDynamic: dynamic, timestamp:datetime)
```

Cree una asignación de JSON para asignar propiedades de registro a las columnas adecuadas. El comando siguiente crea una asignación basada en el ejemplo del paso anterior:

.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"}} 
    ]```

Conceder permisos a la entidad de servicio

Conceda a la entidad de servicio de Crear una entidad de servicio de Microsoft Entra permisos del rol de agente de ingesta de base de datos para trabajar con la base de datos. Para más información, consulte los ejemplos. Reemplace el marcador de posición DatabaseName por el nombre de la base de datos de destino y ApplicationID por el valor AppId que guardó al crear una entidad de servicio de Microsoft Entra.

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

Configuración de Fluent Bit para enviar registros a la tabla

Para configurar Fluent Bit para enviar registros a la tabla en Kusto, cree un archivo de configuración de modo clásico o modo YAML con las siguientes propiedades de salida:

Campo	Descripción	Obligatorio	Predeterminado
Nombre	El nombre de la canalización.		`azure_kusto`
identificador_de_inquilino	El identificador de cliente procede de Crear un principal de servicio en Microsoft Entra.	✔️
ID de cliente	Identificador de aplicación de Crear una entidad de servicio de Microsoft Entra.	✔️
secreto_del_cliente	Valor de la clave secreta del cliente (contraseña) de Crear un principal del servicio de Microsoft Entra.	✔️
managed_identity_client_id	Identificador de cliente de la identidad administrada que se va a usar para la autenticación.	✔️
ingestion_endpoint	Escriba el valor tal como se describe para Ingestion_Endpoint.	✔️
nombre de la base de datos	El nombre de la base de datos que contendrá la tabla de registro.	✔️
nombre_de_tabla	Nombre de la tabla de Crear una tabla de destino.	✔️
ingestion_mapping_reference	Nombre de la asignación de ingesta derivada de Crear una tabla de destino. Si no creó una asignación de ingesta, quite la propiedad del archivo de configuración.
clave_de_registro	Nombre de clave del contenido del registro. Por ejemplo, `log`.		`log`
include_tag_key	Si está habilitada, se anexa una etiqueta a la salida.		`On`
tag_key	El nombre de clave de la etiqueta. Se omite si `include_tag_key` es false.		`tag`
include_time_key	Se anexa una marca de tiempo a la salida, si está habilitada. Usa la `time_key` propiedad .		`On`
time_key	Nombre de clave para la marca de tiempo en los registros. Se omite si `include_time_key` es false.		`timestamp`
ingestion_endpoint_connect_timeout	Tiempo de espera de conexión de varios puntos de conexión de Kusto en segundos.		`60`
compresión_activada	Envía la carga HTTP comprimida (gzip) a Kusto, si está habilitada.		`true`
ingestion_resources_refresh_interval	Intervalo de actualización de recursos de ingesta del punto de conexión de Kusto en segundos.
trabajadores	Número de trabajadores encargados de llevar a cabo las operaciones de vaciado para esta salida.		`0`
buffering_enabled	Si está habilitado, almacena en búfer los datos en el disco antes de ingerirlos en Kusto.		`Off`
ruta_del_buffer	Especifica la ubicación del directorio donde se almacenarán los datos almacenados en búfer si `buffering_enabled` es `On`.		`/tmp/fluent-bit/azure-kusto/`
upload_timeout	Especifica el tiempo de espera de las cargas si `buffering_enabled` es `On`. Los archivos anteriores a esto se ingieren incluso si están por debajo del límite de tamaño.		`30m`
upload_file_size	Especifica el tamaño máximo de un archivo que se va a cargar si `buffering_enabled` es `On`.		`200MB`
azure_kusto_buffer_key	Clave de búfer de Azure Kusto para identificar las instancias del complemento cuando `buffering_enabled` es `On`. Necesario para varias salidas de Azure Kusto con almacenamiento en búfer.		`key`
límite_de_tamaño_del_directorio_de_almacenamiento	El tamaño máximo del directorio donde se almacenan los datos almacenados en búfer si `buffering_enabled` es `On`.		`8GB`
buffer_file_delete_early	Cuando `buffering_enabled` es `On`, si desea eliminar el archivo almacenado en búfer rápidamente después de la creación exitosa del blob.		`Off`
unify_tag	Crea un único archivo de búfer cuando `buffering_enabled` es `On`.		`On`
blob_uri_length	Establezca la longitud del URI del blob generado antes de la ingesta en Kusto.		`64`
scheduler_max_retries	Cuando `buffering_enabled` es `On`, establezca el número máximo de reintentos para la ingesta mediante el programador.		`3`
delete_on_max_upload_error	Cuando `buffering_enabled` es `On`, decidir si eliminar el archivo de búfer en caso de errores máximos de carga.		`Off`
IO_timeout	Configure el tiempo de espera de E/S HTTP para las cargas.		`60s`

Para ver un archivo de configuración de ejemplo, seleccione la pestaña correspondiente:

Modo clásico
Modo 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

Confirmación de la ingesta de datos

Una vez que los datos llegan a la tabla, confirme la transferencia de datos; para ello, compruebe el recuento de filas:
```
FluentBitLogs
| count
```
Para ver un ejemplo de datos de registro, ejecute la consulta siguiente:
```
FluentBitLogs
| take 100
```

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2025-07-16