Plugin de Fluentd

Introducción al producto

Este plugin de salida de Fluentd escribe los datos recopilados en DataHub. Cumple con los estándares de desarrollo de plugins de salida de Fluentd y es fácil de instalar.

Instalación del producto

Instalación mediante RubyGems

Nota: Se recomienda cambiar el origen de RubyGems a https://gems.ruby-china.com.

gem install fluent-plugin-datahub

Instalación local

Fluentd se ejecuta únicamente en un entorno Linux y requiere que Ruby esté instalado.
Hay dos modos de instalación disponibles. Si no ha instalado Fluentd, utilice el modo de instalación con un solo clic para instalar tanto Fluentd como el plugin de DataHub. Si ya cuenta con Fluentd, emplee el modo de instalación independiente para instalar solo el plugin de escritura de DataHub.

1) Instalación con un solo clic: Si no ha instalado Fluentd, descargue el paquete completo de instalación de Fluentd. Nota: El paquete completo de instalación incluye la versión de Fluentd fluentd-0.12.25.gem.

$ tar -xzvf fluentd-with-datahub-0.12.25.tar.gz
$ cd fluentd-with-datahub
$ sudo sh install.sh

2) Instalación independiente: Si ya instaló Fluentd, descargue el paquete del plugin de DataHub para Fluentd. Utilice el comando gem para instalar el plugin de DataHub.

$ sudo gem install --local fluent-plugin-datahub-0.12.25.gem

Casos de uso

Caso de uso 1: Cargar un archivo CSV

A continuación se muestra cómo cargar archivos CSV incrementales a DataHub casi en tiempo real. El archivo CSV tiene el siguiente formato:

0,qe614c760fuk8judu01tn5x055rpt1,true,100.1,14321111111
1,znv1py74o8ynn87k66o32ao4x875wi,true,100.1,14321111111
2,7nm0mtpgo1q0ubuljjjx9b000ybltl,true,100.1,14321111111
3,10t0n6pvonnan16279w848ukko5f6l,true,100.1,14321111111
4,0ub584kw88s6dczd0mta7itmta10jo,true,100.1,14321111111
5,1ltfpf0jt7fhvf0oy4lo8m3z62c940,true,100.1,14321111111
6,zpqsfxqy9379lmcehd7q8kftntrozb,true,100.1,14321111111
7,ce1ga9aln346xcj761c3iytshyzuxg,true,100.1,14321111111
8,k5j2id9a0ko90cykl40s6ojq6gruyi,true,100.1,14321111111
9,ns2zcx9bdip5y0aqd1tdicf7bkdmsm,true,100.1,14321111111
10,54rs9cm1xau2fk66pzyz62tf9tsse4,true,100.1,14321111111

Cada línea representa un registro con campos distintos. El archivo se guarda en la ruta local /temp/test.csv. El tema de DataHub tiene el siguiente esquema:

Nombre del campo	Tipo de campo
id	BIGINT
name	STRING
gender	BOOLEAN
salary	DOUBLE
my_time	TIMESTAMP

Agregue la siguiente configuración de Fluentd a ${CONFIG_HOME}/fluentd_test.conf:

<source>
  @type tail
  path your_file_path
  tag test1
  format csv
  keys id,name,gender,salary,my_time
</source>
<match test1>
  @type datahub
  access_id your_app_id
  access_key your_app_key
  endpoint http://ip:port
  project_name test_project
  topic_name fluentd_performance_test_1
  column_names ["id", "name", "gender", "salary", "my_time"]
  flush_interval 1s
  buffer_chunk_limit 3m
  buffer_queue_limit 128
  dirty_data_continue true
  dirty_data_file path_to_dirty_data_file
  retry_times 3
  put_data_batch_size 1000
</match>

Ejecute el siguiente comando para iniciar Fluentd y enviar datos desde el archivo CSV a DataHub:

${FLUENTD_HOME}/fluentd-with-dataHub/bin/fluentd -c ${CONFIG_HOME}/fluentd_test.conf

Caso de uso 2: Recopilar registros de Log4j

Los registros de Log4j tienen el siguiente formato:

11:48:43.439 [qtp1847995714-17] INFO  AuditInterceptor - [c2un5sh7cu52ek6am1ui1m5h] end /web/v1/project/tefe4mfurtix9kwwyrvfqd0m/node/0m0169kapshvgc3ujskwkk8g/health GET, 4061 ms

Utilice la siguiente configuración de Fluentd:

 <source>
   @type tail
   path bayes.log
   tag test
   format /(?<request_time>\d\d:\d\d:\d\d.\d+)\s+\[(?<thread_id>[\w\-]+)\]\s+(?<log_level>\w+)\s+(?<class>\w+)\s+-\s+\[(?<request_id>\w+)\]\s+(?<detail>.+)/
 </source>
 <match test>
   @type datahub
   access_id your_access_id
   access_key your_access_key
   endpoint http://ip:port
   project_name test_project
   topic_name dataHub_fluentd_out_1
   column_names ["thread_id", "log_level", "class"]
 </match>

Inicie Fluentd con esta configuración para enviar los registros de Log4j a DataHub.

Parámetros de configuración

Input plugin configuration
tag test1                          : Specifies the route. The route is matched against the <match> regular expression.
format csv                         : Collects data in CSV format.
keys id,name,gender,salary,my_time : Specifies the columns to collect. The column names must match the column names in the destination DataHub table.

Output plugin configuration
shard_id 0               : Specifies the shard ID to write to. By default, data is written in a round-robin manner.
shard_keys ["id"]         : Specifies the partition key. The key value is hashed to determine the index of the shard to write to.
flush_interval 1         : Fluentd writes data at least once per second. The default is 60s.
buffer_chunk_limit 3m    : The chunk size. Supports "k" (KB) and "m" (MB). The recommended value is 3m.
buffer_queue_limit 128   : The chunk queue size. This value and buffer_chunk_limit determine the total buffer size.
put_data_batch_size 1000 : Writes data to DataHub for every 1,000 records.
retry_times 3            : The number of retries.
retry_interval  3        : The retry interval in seconds.
dirty_data_continue true : Specifies whether to continue when dirty data is encountered. If set to true, Fluentd retries. After all retries are exhausted, the dirty data is written to a dirty data file.
dirty_data_file /xxx/yyy : Specifies the path to the dirty data file.
column_names ["id"]      : Specifies the columns to collect.

Prueba de rendimiento

Entorno de prueba: Fluentd se ejecutó en un sistema Linux de 2 núcleos y 4 GB. Los resultados de la prueba de rendimiento son los siguientes:

Para registros individuales de 512 B, la velocidad de escritura fue de aproximadamente 2,800 registros/s.
Incrementar put_data_batch_size mejoró ligeramente la velocidad, pero el efecto no fue significativo.
Para registros individuales de 100 KB, el plugin funcionó solo cuando put_data_batch_size se estableció en 100. Los valores de 500 y 1,000 no funcionaron porque el tamaño de los datos para una sola operación de escritura excedió los 50 MB.
La velocidad promedio total de escritura fue de 3 MB/s. Preguntas frecuentes

P: ¿Cómo escribo una expresión regular para el parámetro de formato en Fluentd? R: Utilice este editor de expresiones regulares en línea.