Beats output plugin

Beats output plugin

这是 elastic beats 的输出实现，支持 Filebeat, Metricbeat, Packetbeat, Winlogbeat, Auditbeat, Heartbeat 到 Apache Doris。

该插件用于 beats 输出数据到 Doris，使用 HTTP 协议与 Doris FE Http 接口交互，并通过 Doris 的 stream load 的方式进行数据导入.

了解Doris Stream Load

了解更多关于Doris

兼容性

此插件是使用 Beats 7.3.1 开发和测试的

安装

下载源码

1. mkdir -p $GOPATH/src/github.com/apache/
2. cd $GOPATH/src/github.com/apache/
3. git clone https://github.com/apache/doris
4. cd doris/extension/beats

编译

在 extension/beats/ 目录下执行

1. go build -o filebeat filebeat/filebeat.go
2. go build -o metricbeat metricbeat/metricbeat.go
3. go build -o winlogbeat winlogbeat/winlogbeat.go
4. go build -o packetbeat packetbeat/packetbeat.go
5. go build -o auditbeat auditbeat/auditbeat.go
6. go build -o heartbeat heartbeat/heartbeat.go

使用

您可以使用目录 [./example/] 中的示例配置文件，也可以按照以下步骤创建它。

配置 Beat

添加以下配置到 *beat.yml

1. output.doris:
2. fenodes: ["http://localhost:8030"] # your doris fe address
3. user: root # your doris user
4. password: root # your doris password
5. database: example_db # your doris database
6. table: example_table # your doris table

8. codec_format_string: "%{[message]}" # beat-event format expression to row data
9. headers:
10. column_separator: ","

启动 Beat

使用 filebeat 作为示例

1. ./filebeat/filebeat -c filebeat.yml -e

配置说明

连接 doris 配置:

Name	Description	Default
fenodes	FE 的 HTTP交互地址。例如： [“<http://fe1:8030>“, “<http://fe2:8030>“]
user	用户名，该用户需要有 Doris 对应库表的导入权限
password	密码
database	数据库名
table	表名
labelprefix	导入标识前缀，最终生成的标识为 {label_prefix}{db}{table}{time_stamp}	doris_beats
line_delimiter	用于指定导入数据中的换行符，可以使用做多个字符的组合作为换行符。	\n
headers	用户可以通过 headers 传入 stream-load 导入参数

Beats 配置:

Name	Description	Default
codec_format_string	设置格式化 beats 事件的表达式，格式化结果会作为行数据添加到 http body 中
codec	beats 输出编解码器，格式结果将作为一行添加到 http body 中，优先使用 `codec_format_string`
timeout	设置 http 客户端超时时间
bulk_max_size	批处理的最大事件数	100000
max_retries	发送失败时的最大重试次数，Filebeat 忽略 max_retries 设置并无限期重试。	3
backoff.init	网络错误后尝试重新连接之前等待的秒数	1
backoff.max	网络错误后尝试连接之前等待的最大秒数	60

完整使用示例(Filebeat)

初始化 Doris

1. CREATE DATABASE example_db;

3. CREATE TABLE example_db.example_table (
4. id BIGINT,
5. name VARCHAR(100)
6. )
7. UNIQUE KEY(`id`)
8. DISTRIBUTED BY HASH(`id`) BUCKETS 1
9. PROPERTIES (
10. "replication_num"="1"
11. );

配置 Filebeat

创建 /tmp/beats/filebeat.yml 文件并添加以下配置：

1. filebeat.inputs:
2. - type: log
3. enabled: true
4. paths:
5. - /tmp/beats/example.log

7. output.doris:
8. fenodes: ["http://localhost:8030"] # your doris fe address
9. user: root # your doris user
10. password: root # your doris password
11. database: example_db # your doris database
12. table: example_table # your doris table

14. codec_format_string: "%{[message]}"
15. headers:
16. column_separator: ","

启动 Filebeat

1. ./filebeat/filebeat -c /tmp/beats/filebeat.yml -e

验证数据导入

添加数据到文件 /tmp/beats/example.log

1. echo -e "1,A\n2,B\n3,C\n4,D" >> /tmp/beats/example.log

观察 filebeat 日志，如果没有打印错误日志，则导入成功。这时可以在 example_db.example_table 表中查看导入的数据

常见问题

如何配置批处理提交大小

添加以下内容到您的 *beat.yml 文件中

它表示，如果有 10000 个事件可用或最旧的可用事件已在内存队列中等待 5 秒，此示例配置会将事件批量转发给 doris：

1. queue.mem:
2. events: 10000
3. flush.min_events: 10000
4. flush.timeout: 5s

如何使用其他的 beats(例如 metricbeat)

Doris beats 支持所有的 beats 模块，使用方式参见安装与使用

如何构建 docker 镜像

可以使用安装输出的可执行文件打包 docker 镜像

Beats Doris Output Plugin