如何将ElasticSearch配置为Fluentd的数据源？

可以，Fluentd完全支持将ElasticSearch作为数据源，你可以通过官方维护的elasticsearch输入插件实现ES -> Fluentd -> Kafka的流转流程，具体实现步骤如下：

1. 安装ElasticSearch输入插件 #

首先安装Fluentd的ElasticSearch输入插件：

• 通用Fluentd环境：

gem install fluent-plugin-elasticsearch

• 使用td-agent的环境：

td-agent-gem install fluent-plugin-elasticsearch

2. 配置Fluentd读取ElasticSearch数据 #

在Fluentd的配置文件（如fluent.conf或td-agent.conf）中添加输入段，指定要读取的ES集群和索引信息：

<source>
  @type elasticsearch
  host 你的ES集群地址
  port 你的ES端口（默认9200）
  index_name 目标索引名（支持通配符，如`app-logs-*`）
  scroll_interval 10s
  scroll_size 1000
  tag es.data
  # 若ES开启认证，添加以下配置
  # user ES用户名
  # password ES密码
</source>

关键参数说明：

• scroll_interval/scroll_size：控制批量读取的间隔和单次读取的文档数量，平衡ES负载和数据流转效率
• tag：给读取到的ES数据打标签，用于后续匹配输出规则

3. 配置Fluentd输出到Kafka #

接着添加输出段，将打标后的ES数据投递到指定Kafka主题：

<match es.data>
  @type kafka
  brokers Kafka集群地址列表（逗号分隔，如`kafka1:9092,kafka2:9092`）
  default_topic 目标Kafka主题名
  format json
  # 若Kafka开启认证，添加对应SASL/SSL配置
</match>

4. 验证与启动 #

配置完成后，重启Fluentd服务：

# td-agent环境
systemctl restart td-agent
# 通用Fluentd环境
fluentd -c 你的配置文件路径

检查Fluentd日志确认无连接或配置错误，随后即可查看Kafka主题是否成功接收到来自ES的数据。

注意事项 #

• 确保Fluentd所在节点能同时连通ElasticSearch和Kafka集群
• 针对大型索引，建议结合since_time参数实现增量读取，避免重复消费全量数据
• 根据ES集群负载调整scroll_size参数，防止单次读取数据量过大拖垮集群

内容的提问来源于stack exchange，提问作者MatR