Последнее обновление 19 авг. 2026 г.

Beats

Beats представляет собой агент сбора данных, который поддерживает запись данных в системы хранения через пользовательские плагины вывода. Плагин вывода Beats Doris передаёт данные, собранные Beats, в Doris.

Для записи данных, собранных Filebeat и другими компонентами Beats, в Doris в реальном времени можно использовать плагин вывода Beats Doris. Этот плагин записывает данные через HTTP-интерфейс Doris Stream Load и поддерживает параллельную работу в нескольких потоках, повторные попытки при сбоях, настройку формата и параметров Stream Load, а также вывод информации о скорости записи.

Плагин вывода Beats Doris поддерживает следующие компоненты Beats:

Компонент Beats	Описание
Filebeat	Собирает файлы журналов
Metricbeat	Собирает данные метрик
Packetbeat	Собирает сетевые данные
Winlogbeat	Собирает журналы событий Windows
Auditbeat	Собирает данные аудита
Heartbeat	Собирает данные проверок доступности

Использование плагина вывода Beats Doris обычно включает три шага:

Скачайте или соберите исполняемый файл Beats с плагином вывода в Doris.
Укажите адрес вывода в Doris и другие параметры в файле конфигурации Beats.
Запустите Beats для записи данных в Doris в реальном времени.

Установка

Вы можете скачать исполняемый файл Beats, в который уже включён плагин вывода в Doris, или собрать его из исходного кода.

Загрузка с официального сайта

Скачайте filebeat-doris-2.1.1.

Сборка из исходного кода

В каталоге extension/beats/ исходного кода Doris выполните следующие команды:

cd doris/extension/beats

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

Настройка вывода в Doris

В файле конфигурации Beats настройте вывод в Doris с помощью output.doris. Ниже приведены основные параметры конфигурации:

Параметр конфигурации	Значение по умолчанию	Описание
`fenodes`	Нет	HTTP-адрес Stream Load. Задаётся в виде массива строк; можно указать один или несколько адресов. Например: `["http://fe1:8030", "http://fe2:8030"]`.
`user`	Нет	Имя пользователя Doris. У этого пользователя должны быть права на загрузку в целевую базу данных и таблицу.
`password`	Нет	Пароль пользователя Doris.
`database`	Нет	Имя базы данных Doris для записи.
`table`	Нет	Имя таблицы Doris для записи.
`label_prefix`	`beats`	Префикс Label для Doris Stream Load. Итоговый формат Label: `{label_prefix}_{db}_{table}_{yyyymmdd_hhmmss}_{uuid}`.
`headers`	Нет	Параметр headers для Doris Stream Load. Задаётся в виде словаря YAML.
`codec_format_string`	Нет	Строка формата для данных, передаваемых в Doris Stream Load. Используйте `%{[a][b]}` для обращения к полю `a.b` во входных данных.
`bulk_max_size`	`100000`	Размер пакета для Doris Stream Load.
`max_retries`	`-1`	Количество повторных попыток после сбоя запроса Doris Stream Load. Значение `-1` означает неограниченное количество повторных попыток и используется для обеспечения надёжности передачи данных.
`log_request`	`true`	Определяет, выводить ли в журнал метаданные запросов и ответов Doris Stream Load для диагностики проблем.
`log_progress_interval`	`10`	Интервал вывода информации о скорости записи в журнал, в секундах. Установите значение `0`, чтобы отключить этот вывод.

Сценарий 1: сбор журналов TEXT

В этом сценарии на примере журналов Doris FE показано, как собирать обычные текстовые записи и многострочные трассировки стека и записывать их в Doris.

Шаг 1: подготовка данных

Файлы журналов Doris FE обычно находятся по пути fe/log/fe.log в каталоге установки Doris. Журнал FE представляет собой типичный журнал Java-программы и содержит такие поля, как временная метка, уровень журналирования, имя потока, место в коде и содержимое записи. В журнале есть как обычные записи, так и записи об исключениях с трассировками стека. Поскольку трассировка стека занимает несколько строк, при сборе данных основную запись журнала и соответствующую трассировку стека необходимо объединить в одну запись.

2024-07-08 21:18:01,432 INFO (Statistics Job Appender|61) [StatisticsJobAppender.runAfterCatalogReady():70] Stats table not available, skip
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}
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
        at org.apache.doris.planner.OlapScanNode.addScanRangeLocations(OlapScanNode.java:931) ~[doris-fe.jar:1.2-SNAPSHOT]
        at org.apache.doris.planner.OlapScanNode.computeTabletInfo(OlapScanNode.java:1197) ~[doris-fe.jar:1.2-SNAPSHOT]

Шаг 2: создание таблицы Doris

Схема таблицы включает такие поля, как время создания записи журнала, время её сбора, имя хоста, путь к файлу журнала, тип журнала, уровень журналирования, имя потока, место в коде и содержимое записи.

CREATE TABLE `doris_log` (
  `log_time` datetime NULL COMMENT 'log content time',
  `collect_time` datetime NULL COMMENT 'log agent collect time',
  `host` text NULL COMMENT 'hostname or ip',
  `path` text NULL COMMENT 'log file path',
  `type` text NULL COMMENT 'log type',
  `level` text NULL COMMENT 'log level',
  `thread` text NULL COMMENT 'log thread',
  `position` text NULL COMMENT 'log code position',
  `message` text NULL COMMENT 'log message',
  INDEX idx_host (`host`) USING INVERTED COMMENT '',
  INDEX idx_path (`path`) USING INVERTED COMMENT '',
  INDEX idx_type (`type`) USING INVERTED COMMENT '',
  INDEX idx_level (`level`) USING INVERTED COMMENT '',
  INDEX idx_thread (`thread`) USING INVERTED COMMENT '',
  INDEX idx_position (`position`) USING INVERTED COMMENT '',
  INDEX idx_message (`message`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true") COMMENT ''
) ENGINE=OLAP
DUPLICATE KEY(`log_time`)
COMMENT 'OLAP'
PARTITION BY RANGE(`log_time`) ()
DISTRIBUTED BY RANDOM BUCKETS 10
PROPERTIES (
"replication_num" = "1",
"dynamic_partition.enable" = "true",
"dynamic_partition.time_unit" = "DAY",
"dynamic_partition.start" = "-7",
"dynamic_partition.end" = "1",
"dynamic_partition.prefix" = "p",
"dynamic_partition.buckets" = "10",
"dynamic_partition.create_history_partition" = "true",
"compaction_policy" = "time_series"
);

Шаг 3: настройка Filebeat

Файл конфигурации сбора журналов Filebeat (например, filebeat_doris_log.yml) использует формат YAML и состоит из четырёх основных частей:

input: читает исходный файл журнала.
processors: преобразует и разбирает содержимое журнала.
queue.mem: настраивает внутреннюю буферную очередь Filebeat.
output: выводит данные в Doris.

# 1. input is responsible for reading raw data.
# type: log is the log input plugin, which lets you configure the path of the log file to read.
# multiline appends lines that do not start with a timestamp to the previous line, used to merge stack traces with the main log entry.
# The log input stores the log content in the message field, and also generates metadata fields such as agent.host and log.file.path.
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /path/to/your/log
  multiline:
    type: pattern
    # A line starting with yyyy-mm-dd HH:MM:SS is recognized as a new log entry; other lines are appended to the previous entry.
    pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
    negate: true
    match: after
    skip_newline: true

# 2. processors handle data transformation.
processors:
# Use the js script plugin to replace \t in the log with spaces, to avoid JSON parsing errors.
- script:
    lang: javascript
    source: >
        function process(event) {
            var msg = event.Get("message");
            msg = msg.replace(/\t/g, "  ");
            event.Put("message", msg);
        }
# Use the dissect plugin for simple log parsing.
- dissect:
    # 2024-06-08 18:26:25,481 INFO (report-thread|199) [ReportHandler.cpuReport():617] begin to handle
    tokenizer: "%{day} %{time} %{log_level} (%{thread}) [%{position}] %{content}"
    target_prefix: ""
    ignore_failure: true
    overwrite_keys: true

# 3. queue.mem configures the total number of events in the internal buffer queue, the flush batch size, and the flush time interval.
queue.mem:
  events: 1000000
  flush.min_events: 100000
  flush.timeout: 10s

# 4. output is responsible for sending data to Doris.
# doris output uses the Stream Load HTTP interface to write to Doris.
# headers specifies the Stream Load data format as JSON.
# codec_format_string formats the output content in a printf-like manner.
# In the example, you can reference Filebeat built-in fields (such as agent.hostname) or fields generated by processors (such as day).
output.doris:
  fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
  user: "your_username"
  password: "your_password"
  database: "your_db"
  table: "your_table"
  # output string format
  ## %{[agent][hostname]} and %{[log][file][path]} are Filebeat built-in metadata.
  ## Other commonly used Filebeat metadata includes the collection timestamp %{[@timestamp]}.
  ## %{[day]} and %{[time]} are fields obtained from the dissect parsing above.
  codec_format_string: '{"ts": "%{[day]} %{[time]}", "host": "%{[agent][hostname]}", "path": "%{[log][file][path]}", "message": "%{[message]}"}'
  headers:
    format: "json"
    read_json_by_line: "true"
    load_to_single_tablet: "true"

Шаг 4: запуск Filebeat

Выполните следующую команду для запуска Filebeat:

./filebeat-doris -f config/filebeat_doris_log.yml

Если log_request имеет значение true, в журнал выводятся параметры запроса и результат ответа для каждого вызова Stream Load.

doris stream load response:
{
    "TxnId": 45464,
    "Label": "beats_log_db_doris_log_20240708_223532_539_6c20a0d1-dcab-4b8e-9bc0-76b46a929bd1",
    "Comment": "",
    "TwoPhaseCommit": "false",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 452,
    "NumberLoadedRows": 452,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 277230,
    "LoadTimeMs": 1797,
    "BeginTxnTimeMs": 0,
    "StreamLoadPutTimeMs": 18,
    "ReadDataTimeMs": 9,
    "WriteDataTimeMs": 1758,
    "CommitAndPublishTimeMs": 18
}

По умолчанию информация о скорости записи выводится в журнал каждые 10 секунд. Она включает объём данных с момента запуска (в MB и ROWS), общую скорость (в MB/s и R/s) и скорость за последние 10 секунд.

total 11 MB 18978 ROWS, total speed 0 MB/s 632 R/s, last 10 seconds speed 1 MB/s 1897 R/s

Сценарий 2: сбор журналов JSON

В этом сценарии на примере данных GitHub Events Archive показано, как собирать журналы JSON и записывать их в Doris.

Шаг 1: подготовка данных

GitHub Events Archive содержит архивные данные о событиях, связанных с действиями пользователей GitHub, в формате JSON. Эти данные можно скачать с сайта GitHub Archive. Например, чтобы скачать данные за 15:00 1 января 2024 года:

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

Ниже приведён пример записи. В исходных данных каждая строка содержит одну запись JSON; здесь запись отформатирована для удобства чтения.

{
  "id": "37066529221",
  "type": "PushEvent",
  "actor": {
    "id": 46139131,
    "login": "Bard89",
    "display_login": "Bard89",
    "gravatar_id": "",
    "url": "https://api.github.com/users/Bard89",
    "avatar_url": "https://avatars.githubusercontent.com/u/46139131?"
  },
  "repo": {
    "id": 780125623,
    "name": "Bard89/talk-to-me",
    "url": "https://api.github.com/repos/Bard89/talk-to-me"
  },
  "payload": {
    "repository_id": 780125623,
    "push_id": 17799451992,
    "size": 1,
    "distinct_size": 1,
    "ref": "refs/heads/add_mvcs",
    "head": "f03baa2de66f88f5f1754ce3fa30972667f87e81",
    "before": "85e6544ede4ae3f132fe2f5f1ce0ce35a3169d21"
  },
  "public": true,
  "created_at": "2024-04-01T23:00:00Z"
}

Шаг 2: создание таблицы Doris

CREATE DATABASE log_db;
USE log_db;

CREATE TABLE github_events
(
  `created_at` DATETIME,
  `id` BIGINT,
  `type` TEXT,
  `public` BOOLEAN,
  `actor` VARIANT,
  `repo` VARIANT,
  `payload` TEXT,
  INDEX `idx_id` (`id`) USING INVERTED,
  INDEX `idx_type` (`type`) USING INVERTED,
  INDEX `idx_actor` (`actor`) USING INVERTED,
  INDEX `idx_host` (`repo`) USING INVERTED,
  INDEX `idx_payload` (`payload`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true")
)
ENGINE = OLAP
DUPLICATE KEY(`created_at`)
PARTITION BY RANGE(`created_at`) ()
DISTRIBUTED BY RANDOM BUCKETS 10
PROPERTIES (
"replication_num" = "1",
"compaction_policy" = "time_series",
"dynamic_partition.enable" = "true",
"dynamic_partition.create_history_partition" = "true",
"dynamic_partition.time_unit" = "DAY",
"dynamic_partition.start" = "-30",
"dynamic_partition.end" = "1",
"dynamic_partition.prefix" = "p",
"dynamic_partition.buckets" = "10",
"dynamic_partition.replication_num" = "1"
);

Шаг 3: настройка Filebeat

По сравнению с конфигурацией сбора журналов TEXT сценарий JSON имеет два отличия:

processors не используется, поскольку дополнительные преобразования не нужны.
codec_format_string напрямую выводит всё содержимое message, то есть исходное содержимое JSON.

# input
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /path/to/your/log

# queue and batch
queue.mem:
  events: 1000000
  flush.min_events: 100000
  flush.timeout: 10s

# output
output.doris:
  fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
  user: "your_username"
  password: "your_password"
  database: "your_db"
  table: "your_table"
  # output string format
  ## Output the message of each line in the original file as-is.
  ## headers specifies format: "json", so Stream Load automatically parses JSON fields and writes them into the corresponding fields of the Doris table.
  codec_format_string: '%{[message]}'
  headers:
    format: "json"
    read_json_by_line: "true"
    load_to_single_tablet: "true"

Шаг 4: запуск Filebeat

Выполните следующую команду для запуска Filebeat:

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

Установка​

Загрузка с официального сайта​

Сборка из исходного кода​

Настройка вывода в Doris​

Сценарий 1: сбор журналов TEXT​

Шаг 1: подготовка данных​

Шаг 2: создание таблицы Doris​

Шаг 3: настройка Filebeat​

Шаг 4: запуск Filebeat​

Сценарий 2: сбор журналов JSON​

Шаг 1: подготовка данных​

Шаг 2: создание таблицы Doris​

Шаг 3: настройка Filebeat​

Шаг 4: запуск Filebeat​

Установка

Загрузка с официального сайта

Сборка из исходного кода

Настройка вывода в Doris

Сценарий 1: сбор журналов TEXT

Шаг 1: подготовка данных

Шаг 2: создание таблицы Doris

Шаг 3: настройка Filebeat

Шаг 4: запуск Filebeat

Сценарий 2: сбор журналов JSON

Шаг 1: подготовка данных

Шаг 2: создание таблицы Doris

Шаг 3: настройка Filebeat

Шаг 4: запуск Filebeat