如何在Terraform与Elasticsearch Sink Connector中配置自定义Routing？

方案1：直接通过Elasticsearch Sink Connector配置实现 #

ES Sink Connector v14.0.2支持直接指定_routing字段，无需额外的Kafka Stream处理，核心配置如下：

在连接器配置中添加以下关键参数：

# 启用自定义路由功能
routing.enabled=true
# 指定消息中的嵌套字段作为路由值，嵌套字段用点分隔路径
routing.field=joinField.parent
# 当路由字段不存在时的 fallback 策略，选ignore避免写入失败
routing.missing=ignore

配置后，连接器会自动提取消息里的joinField.parent字段值，作为Elasticsearch文档的_routing属性。

方案2：通过Kafka Stream预处理消息（适合需修改/验证路由字段的场景） #

如果需要对joinField.parent做额外处理（比如为空时补默认值、格式转换），可以在Kafka Stream中添加逻辑，将路由值转存到顶层字段：

1. Kafka Stream处理逻辑示例（以Java为例，可根据实际语言调整）：

KStream<String, JsonNode> sourceStream = builder.stream("cash-box-dynamic");
KStream<String, JsonNode> processedStream = sourceStream.mapValues(value -> {
    JsonNode parentValue = value.path("joinField").path("parent");
    if (!parentValue.isMissingNode()) {
        ((ObjectNode) value).put("es_routing", parentValue.asText());
    }
    return value;
});
processedStream.to("cash-box-dynamic-processed");

2. 同步调整ES Sink Connector的路由配置：

routing.enabled=true
routing.field=es_routing
routing.missing=ignore

Terraform管理Connector配置示例 #

将上述路由配置写入Terraform的kafka_connector资源块中：

resource "kafka_connector" "es_sink_cash_box" {
  name = "elasticsearch-sink-cash-box-dynamic"
  config = {
    "connector.class" = "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"
    "tasks.max" = "1"
    "topics" = "cash-box-dynamic"
    "key.ignore" = "true"
    "connection.url" = "http://elasticsearch:9200"
    "index.name" = "cash-box-dynamic"
    # 路由核心配置
    "routing.enabled" = "true"
    "routing.field" = "joinField.parent"
    "routing.missing" = "ignore"
    # 其他已有配置（如格式转换、认证等）
    "value.converter" = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
  }
}

注意事项 #

• 确保Elasticsearch索引模板已正确配置_routing相关规则（Parent-Join类型索引通常需开启routing.required）
• 测试时可发送样例消息，验证ES文档的_routing字段是否匹配预期值
• v14版本的Sink Connector支持点分隔的嵌套字段路径，无需额外处理

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