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 开发和测试的

安装

下载源码

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

编译

在 extension/beats/ 目录下执行

go build -o filebeat filebeat/filebeat.go
go build -o metricbeat metricbeat/metricbeat.go
go build -o winlogbeat winlogbeat/winlogbeat.go
go build -o packetbeat packetbeat/packetbeat.go
go build -o auditbeat auditbeat/auditbeat.go
go build -o heartbeat heartbeat/heartbeat.go

您将在各个子目录目录下得到可执行文件

使用

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

配置 Beat

添加以下配置到 *beat.yml

output.doris:
  fenodes: ["http://localhost:8030"] # your doris fe address
  user: root # your doris user
  password: root # your doris password
  database: example_db # your doris database
  table: example_table # your doris table
  codec_format_string: "%{[message]}" # beat-event format expression to row data
  headers:
    column_separator: ","

启动 Beat

使用 filebeat 作为示例

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

配置说明

连接 doris 配置:

Beats 配置:

完整使用示例(Filebeat)

初始化 Doris

CREATE DATABASE example_db;
CREATE TABLE example_db.example_table (
    id BIGINT,
    name VARCHAR(100)
)
UNIQUE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
    "replication_num"="1"
);

配置 Filebeat

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

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /tmp/beats/example.log
output.doris:
  fenodes: ["http://localhost:8030"] # your doris fe address
  user: root # your doris user
  password: root # your doris password
  database: example_db # your doris database
  table: example_table # your doris table
  codec_format_string: "%{[message]}"
  headers:
    column_separator: ","

启动 Filebeat

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

验证数据导入

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

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

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

更多配置示例

指定导入的 columns

创建 /tmp/beats/example.log 文件并添加以下内容:

1,A
2,B

配置 columns

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /tmp/beats/example.log
output.doris:
  ...
  codec_format_string: "%{[message]}"
  headers:
    columns: "id,name"

采集 json 文件

创建 /tmp/beats/example.json 文件并添加以下内容:

{"id":  1, "name": "A"}
{"id":  2, "name": "B"}

配置 headers

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /tmp/beats/example.json
output.doris:
  ...
  codec_format_string: "%{[message]}"
  headers:
    format: json
    read_json_by_line: true

编码输出字段

创建 /tmp/beats/example.log 文件并添加以下内容:

1,A
2,B

配置 codec_format_string

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /tmp/beats/example.log
output.doris:
  ...
  codec_format_string: "%{[message]},%{[@timestamp]},%{[@metadata.type]}"
  headers:
    columns: "id,name,beat_timestamp,beat_metadata_type"

常见问题

如何配置批处理提交大小

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

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

queue.mem:
  events: 10000
  flush.min_events: 10000
  flush.timeout: 5s

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

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

如何构建 docker 镜像

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