Beats Doris output plugin

Beats 是一个数据采集 Agent，它支持自定义输出插件将数据写入存储系统，Beats Doris output plugin 是输出到 Doris 的插件。

Beats Doris output plugin 支持 Filebeat, Metricbeat, Packetbeat, Winlogbeat, Auditbeat, Heartbeat 。

Beats Doris output plugin 调用 Doris Stream Load HTTP 接口将数据实时写入 Doris，提供多线程并发，失败重试，自定义 Stream Load 格式和参数，输出写入速度等能力。

使用 Beats Doris output plugin 主要有三个步骤：

  1. 下载或编译包含 Doris output plugin 的 Beats二进制程序
  2. 配置 Beats 输出地址和其他参数
  3. 启动 Beats 将数据实时写入 Doris

安装

从官网下载

https://apache-doris-releases.oss-accelerate.aliyuncs.com/filebeat-doris-2.0.0

从源码编译

在 extension/beats/ 目录下执行

  1. cd doris/extension/beats
  3. go build -o filebeat-doris filebeat/filebeat.go
  4. go build -o metricbeat-doris metricbeat/metricbeat.go
  5. go build -o winlogbeat-doris winlogbeat/winlogbeat.go
  6. go build -o packetbeat-doris packetbeat/packetbeat.go
  7. go build -o auditbeat-doris auditbeat/auditbeat.go
  8. go build -o heartbeat-doris heartbeat/heartbeat.go

参数配置

Beats Doris output plugin 的配置如下：

配置说明
http_hostsStream Load HTTP 地址，格式是字符串数组，可以有一个或者多个元素，每个元素是 host:port。 例如：[“http://fe1:8030“, “http://fe2:8030“]
userDoris 用户名，该用户需要有doris对应库表的导入权限
passwordDoris 用户的密码
database要写入的 Doris 库名
table要写入的 Doris 表名
label_prefixDoris Stream Load Label 前缀，最终生成的 Label 为 {labelprefix}{db}{table}{yyyymmddhhmmss}{uuid} ，默认值是 beats
headersDoris Stream Load 的 headers 参数，语法格式为 YAML map
codec_format_string输出到 Doris Stream Load 的format string，%{[a][b]} 代表输入中的 a.b 字段，参考后续章节的使用示例
bulk_max_sizeDoris Stream Load 的 batch size，默认为100000
max_retriesDoris Stream Load 请求失败重试次数，默认为 -1 无限重试保证数据可靠性
log_request日志中是否输出 Doris Stream Load 请求和响应元数据，用于排查问题，默认为 true
log_progress_interval日志中输出速度的时间间隔，单位是秒，默认为 10，设置为 0 可以关闭这种日志

使用示例

TEXT 日志采集示例

该示例以 Doris FE 的日志为例展示 TEXT 日志采集。

1. 数据

FE 日志文件一般位于 Doris 安装目录下的 fe/log/fe.log 文件，是典型的 Java 程序日志，包括时间戳，日志级别，线程名，代码位置，日志内容等字段。不仅有正常的日志，还有带 stacktrace 的异常日志，stacktrace 是跨行的，日志采集存储需要把主日志和 stacktrace 组合成一条日志。

  1. 2024-07-08 21:18:01,432 INFO (Statistics Job Appender|61) [StatisticsJobAppender.runAfterCatalogReady():70] Stats table not available, skip
  2. 2024-07-08 21:18:53,710 WARN (STATS_FETCH-0|208) [StmtExecutor.executeInternalQuery():3332] Failed to run internal SQL: OriginStatement{originStmt='SELECT * FROM __internal_schema.column_statistics WHERE part_id is NULL  ORDER BY update_time DESC LIMIT 500000', idx=0}
  3. org.apache.doris.common.UserException: errCode = 2, detailMessage = tablet 10031 has no queryable replicas. err: replica 10032's backend 10008 does not exist or not alive
  4.         at org.apache.doris.planner.OlapScanNode.addScanRangeLocations(OlapScanNode.java:931) ~[doris-fe.jar:1.2-SNAPSHOT]
  5.         at org.apache.doris.planner.OlapScanNode.computeTabletInfo(OlapScanNode.java:1197) ~[doris-fe.jar:1.2-SNAPSHOT]

2. 建表

表结构包括日志的产生时间，采集时间，主机名，日志文件路径，日志类型，日志级别，线程名，代码位置，日志内容等字段。

  1. CREATE TABLE `doris_log` (
  2.   `log_time` datetime NULL COMMENT 'log content time',
  3.   `collect_time` datetime NULL COMMENT 'log agent collect time',
  4.   `host` text NULL COMMENT 'hostname or ip',
  5.   `path` text NULL COMMENT 'log file path',
  6.   `type` text NULL COMMENT 'log type',
  7.   `level` text NULL COMMENT 'log level',
  8.   `thread` text NULL COMMENT 'log thread',
  9.   `position` text NULL COMMENT 'log code position',
  10.   `message` text NULL COMMENT 'log message',
  11.   INDEX idx_host (`host`) USING INVERTED COMMENT '',
  12.   INDEX idx_path (`path`) USING INVERTED COMMENT '',
  13.   INDEX idx_type (`type`) USING INVERTED COMMENT '',
  14.   INDEX idx_level (`level`) USING INVERTED COMMENT '',
  15.   INDEX idx_thread (`thread`) USING INVERTED COMMENT '',
  16.   INDEX idx_position (`position`) USING INVERTED COMMENT '',
  17.   INDEX idx_message (`message`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true") COMMENT ''
  18. ) ENGINE=OLAP
  19. DUPLICATE KEY(`log_time`)
  20. COMMENT 'OLAP'
  21. PARTITION BY RANGE(`log_time`) ()
  22. DISTRIBUTED BY RANDOM BUCKETS 10
  23. PROPERTIES (
  24. "replication_num" = "1",
  25. "dynamic_partition.enable" = "true",
  26. "dynamic_partition.time_unit" = "DAY",
  27. "dynamic_partition.start" = "-7",
  28. "dynamic_partition.end" = "1",
  29. "dynamic_partition.prefix" = "p",
  30. "dynamic_partition.buckets" = "10",
  31. "dynamic_partition.create_history_partition" = "true",
  32. "compaction_policy" = "time_series"
  33. );

3. 配置

filebeat 日志采集的配置文件如 filebeat_doris_log.yml 是 YAML 格式，主要由 4 部分组成，分别对应 ETL 的各个部分：

  1. input 负责读取原始数据
  2. processor 负责做数据转换
  3. queue.mem 配置 filebeat 内部的缓冲队列
  4. output 负责将数据输出
  1. # 1. input 负责读取原始数据
  2. # type: log 是一个 log input plugin，可以配置读取的日志文件路径，通过 multiline 功能将非时间开头的行拼接到上一行后面，实现 stacktrace 和主日志合并的效果。log input 会将日志内容保存在 message 字段中，另外还有一些元数据字段比如 agent.host，log.file.path。
  3. filebeat.inputs:
  4. - type: log
  5.   enabled: true
  6.   paths:
  7.     - /path/to/your/log
  8.   # multiline 可以将跨行的日志（比如Java stacktrace）拼接起来
  9.   multiline:
  10.     type: pattern
  11.     # 效果：以 yyyy-mm-dd HH:MM:SS 开头的行认为是一条新的日志，其他都拼接到上一条日志
  12.     pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
  13.     negate: true
  14.     match: after
  15.     skip_newline: true
  17. # 2. processors 部分负责数据转换
  18. processors:
  19. # 用 js script 插件将日志中的 \t 替换成空格，避免JSON解析报错
  20. - script:
  21.     lang: javascript
  22.     source: >
  23.         function process(event) {
  24.             var msg = event.Get("message");
  25.             msg = msg.replace(/\t/g, "  ");
  26.             event.Put("message", msg);
  27.         }
  28. # 用 dissect 插件做简单的日志解析
  29. - dissect:
  30.     # 2024-06-08 18:26:25,481 INFO (report-thread|199) [ReportHandler.cpuReport():617] begin to handle
  31.     tokenizer: "%{day} %{time} %{log_level} (%{thread}) [%{position}] %{content}"
  32.     target_prefix: ""
  33.     ignore_failure: true
  34.     overwrite_keys: true
  36. # 3. 内部的缓冲队列总条数，flush batch 条数，flush 时间间隔
  37. queue.mem:
  38.   events: 1000000
  39.   flush.min_events: 100000
  40.   flush.timeout: 10s
  42. # 4. output 部分负责数据输出
  43. # doris output 将数据输出到 Doris，使用的是 Stream Load HTTP 接口。通过 headers 参数指定了 Stream Load 的数据格式为 JSON，通过 codec_format_string 参数用类似 printf 的方式格式化输出到 Doris 的数据。比如下面的例子基于 filebeat 内部的字段 format 出一个 JSON，这些字段可以是 filebeat 内置字段如 agent.hostname，也可以是 processor 比如 dissect 生产的字段如 day，通过 %{[a][b]} 的方式引用，，Stream Load 会自动将 JSON 字段写入对应的 Doris 表的字段。
  44. output.doris:
  45.   fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
  46.   user: "your_username"
  47.   password: "your_password"
  48.   database: "your_db"
  49.   table: "your_table"
  50.   # output string format
  51.   ## %{[agent][hostname]} %{[log][file][path]} 是filebeat自带的metadata
  52.   ## 常用的 filebeat metadata 还是有采集时间戳 %{[@timestamp]}
  53.   ## %{[day]} %{[time]} 是上面 dissect 解析得到字段
  54.   codec_format_string: '{"ts": "%{[day]} %{[time]}", "host": "%{[agent][hostname]}", "path": "%{[log][file][path]}", "message": "%{[message]}"}'
  55.   headers:
  56.     format: "json"
  57.     read_json_by_line: "true"
  58.     load_to_single_tablet: "true"

4. 运行 Filebeat

  2. ./filebeat-doris -f config/filebeat_doris_log.yml
  4. # log_request 为 true 时日志会输出每次 Stream Load 的请求参数和响应结果
  6. doris stream load response:
  7. {
  8.     "TxnId": 45464,
  9.     "Label": "logstash_log_db_doris_log_20240708_223532_539_6c20a0d1-dcab-4b8e-9bc0-76b46a929bd1",
  10.     "Comment": "",
  11.     "TwoPhaseCommit": "false",
  12.     "Status": "Success",
  13.     "Message": "OK",
  14.     "NumberTotalRows": 452,
  15.     "NumberLoadedRows": 452,
  16.     "NumberFilteredRows": 0,
  17.     "NumberUnselectedRows": 0,
  18.     "LoadBytes": 277230,
  19.     "LoadTimeMs": 1797,
  20.     "BeginTxnTimeMs": 0,
  21.     "StreamLoadPutTimeMs": 18,
  22.     "ReadDataTimeMs": 9,
  23.     "WriteDataTimeMs": 1758,
  24.     "CommitAndPublishTimeMs": 18
  25. }
  27. # 默认每隔 10s 会日志输出速度信息，包括自启动以来的数据量（MB 和 ROWS），总速度（MB/s 和 R/S），最近 10s 速度
  28. total 11 MB 18978 ROWS, total speed 0 MB/s 632 R/s, last 10 seconds speed 1 MB/s 1897 R/s

JSON 日志采集示例

该样例以 github events archive 的数据为例展示 JSON 日志采集。

1. 数据

github events archive 是 github 用户操作事件的归档数据，格式是 JSON，可以从 https://www.gharchive.org/ 下载，比如下载 2024年1月1日15点的数据。

  1. wget https://data.gharchive.org/2024-01-01-15.json.gz

下面是一条数据样例，实际一条数据一行，这里为了方便展示进行了格式化。

  1. {
  2.   "id": "37066529221",
  3.   "type": "PushEvent",
  4.   "actor": {
  5.     "id": 46139131,
  6.     "login": "Bard89",
  7.     "display_login": "Bard89",
  8.     "gravatar_id": "",
  9.     "url": "https://api.github.com/users/Bard89",
  10.     "avatar_url": "https://avatars.githubusercontent.com/u/46139131?"
  11.   },
  12.   "repo": {
  13.     "id": 780125623,
  14.     "name": "Bard89/talk-to-me",
  15.     "url": "https://api.github.com/repos/Bard89/talk-to-me"
  16.   },
  17.   "payload": {
  18.     "repository_id": 780125623,
  19.     "push_id": 17799451992,
  20.     "size": 1,
  21.     "distinct_size": 1,
  22.     "ref": "refs/heads/add_mvcs",
  23.     "head": "f03baa2de66f88f5f1754ce3fa30972667f87e81",
  24.     "before": "85e6544ede4ae3f132fe2f5f1ce0ce35a3169d21"
  25.   },
  26.   "public": true,
  27.   "created_at": "2024-04-01T23:00:00Z"
  28. }

2. Doris 建表

  1. CREATE DATABASE log_db;
  2. USE log_db;
  5. CREATE TABLE github_events
  6. (
  7.   `created_at` DATETIME,
  8.   `id` BIGINT,
  9.   `type` TEXT,
  10.   `public` BOOLEAN,
  11.   `actor` VARIANT,
  12.   `repo` VARIANT,
  13.   `payload` TEXT,
  14.   INDEX `idx_id` (`id`) USING INVERTED,
  15.   INDEX `idx_type` (`type`) USING INVERTED,
  16.   INDEX `idx_actor` (`actor`) USING INVERTED,
  17.   INDEX `idx_host` (`repo`) USING INVERTED,
  18.   INDEX `idx_payload` (`payload`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true")
  19. )
  20. ENGINE = OLAP
  21. DUPLICATE KEY(`created_at`)
  22. PARTITION BY RANGE(`created_at`) ()
  23. DISTRIBUTED BY RANDOM BUCKETS 10
  24. PROPERTIES (
  25. "replication_num" = "1",
  26. "compaction_policy" = "time_series",
  27. "enable_single_replica_compaction" = "true",
  28. "dynamic_partition.enable" = "true",
  29. "dynamic_partition.create_history_partition" = "true",
  30. "dynamic_partition.time_unit" = "DAY",
  31. "dynamic_partition.start" = "-30",
  32. "dynamic_partition.end" = "1",
  33. "dynamic_partition.prefix" = "p",
  34. "dynamic_partition.buckets" = "10",
  35. "dynamic_partition.replication_num" = "1"
  36. );

3. Filebeat 配置

这个配置文件和之前 TEXT 日志采集不同的有下面几点：

  1. 没有用 processors，因为不需要额外的处理转换
  2. output 中的 codec_format_string 很简单，直接输出整个 message，也就是原始内容
  1. # input
  2. filebeat.inputs:
  3. - type: log
  4.   enabled: true
  5.   paths:
  6.     - /path/to/your/log
  8. # queue and batch
  9. queue.mem:
  10.   events: 1000000
  11.   flush.min_events: 100000
  12.   flush.timeout: 10s
  14. # output
  15. output.doris:
  16.   fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
  17.   user: "your_username"
  18.   password: "your_password"
  19.   database: "your_db"
  20.   table: "your_table"
  21.   # output string format
  22.   ## 直接把原始文件每一行的 message 原样输出，由于 headers 指定了 format: "json"，Stream Load 会自动解析 JSON 字段写入对应的 Doris 表的字段。
  23.   codec_format_string: '%{[message]}'
  24.   headers:
  25.     format: "json"
  26.     read_json_by_line: "true"
  27.     load_to_single_tablet: "true"

4. 运行 Filebeat

  1. ./filebeat-doris -f config/filebeat_github_events.yml