如何实现Elasticsearch→Kafka→InfluxDB数据流转？Confluent与Kafka Connect疑问

1. 如何使用Confluent Platform实现Elasticsearch到Kafka的数据发送？ #

步骤1：确认环境连通与版本兼容 #

• 确保Confluent Connect节点能访问Elasticsearch集群的9200端口，可通过curl http://<ES_HOST>:9200测试连通性
• 保证Elasticsearch源连接器版本与Elasticsearch主版本匹配（比如Confluent 7.x对应ES 7.x/8.x）

步骤2：配置并部署Elasticsearch源连接器 #

准备连接器配置文件（示例为elasticsearch-source.properties）：

name=elasticsearch-source-connector
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSourceConnector
tasks.max=1
elasticsearch.hosts=http://<ES_HOST>:9200
# 若ES开启认证，添加以下两行
elasticsearch.username=<ES_USER>
elasticsearch.password=<ES_PASS>
index.patterns=your_target_index*
topic.prefix=es_
# 选择同步模式：incrementing（自增字段）/timestamp（时间字段）/bulk（全量）
mode=incrementing
incrementing.field.name=id

关键参数说明：

• index.patterns：指定要同步的Elasticsearch索引（支持通配符）
• topic.prefix：同步数据会发送到前缀+索引名的Kafka主题
• mode：根据业务场景选择同步模式，需确保对应字段存在且符合要求

部署连接器可通过Confluent CLI：

confluent connect cluster create --config elasticsearch-source.properties

或REST API：

curl -X POST -H "Content-Type: application/json" http://<CONNECT_HOST>:8083/connectors -d '{
  "name": "elasticsearch-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSourceConnector",
    "tasks.max": "1",
    "elasticsearch.hosts": "http://<ES_HOST>:9200",
    "index.patterns": "your_target_index*",
    "topic.prefix": "es_",
    "mode": "incrementing",
    "incrementing.field.name": "id"
  }
}'

步骤3：排查同步卡住的问题 #

• 查看Connect节点日志（默认路径/var/log/confluent/connect/connect.log），定位报错原因（如ES索引权限不足、同步字段不存在、Kafka主题未创建）
• 验证Kafka主题自动创建功能是否开启（auto.create.topics.enable=true），或手动提前创建对应主题
• 检查同步模式配置是否匹配ES数据结构（比如mode=incrementing时，索引需有唯一自增的id字段）

步骤4：解决连接器配置丢失问题 #

Confluent Connect默认内存模式会导致重启后配置丢失，需切换为分布式模式并配置持久化存储：

1. 修改Connect配置文件connect-distributed.properties：

config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

2. 提前创建这三个Kafka主题（需设置足够副本数，建议生产环境副本数≥3）：

kafka-topics --create --topic connect-configs --bootstrap-server <KAFKA_HOST>:9092 --replication-factor 3 --partitions 1 --config cleanup.policy=compact
kafka-topics --create --topic connect-offsets --bootstrap-server <KAFKA_HOST>:9092 --replication-factor 3 --partitions 25 --config cleanup.policy=compact
kafka-topics --create --topic connect-status --bootstrap-server <KAFKA_HOST>:9092 --replication-factor 3 --partitions 5 --config cleanup.policy=compact

3. 重启Connect服务，后续配置会持久化到Kafka主题中，关机重启后不会丢失

2. 使用Kafka Connect是否必须依赖Confluent Platform？ #

不需要。Kafka Connect是Apache Kafka的核心组件，属于Apache Kafka官方项目的一部分，无需依赖Confluent Platform即可使用：

• 直接使用Apache Kafka自带的Kafka Connect，搭配社区开源的连接器（比如Elasticsearch源连接器、InfluxDB下沉连接器均有Apache官方或社区维护的版本）
• Confluent Platform提供的是增强版工具链（如Confluent CLI、监控面板、预封装的认证组件）和官方支持的连接器，而Apache Kafka原生Connect是基础版本，需自行管理连接器的下载与配置

内容的提问来源于stack exchange，提问作者Sarindra Thérèse