如何在KSQLDB中构造Elasticsearch父子关联数据？

1. 确认Elasticsearch的Join字段映射 #

先确保你已在ES索引中正确定义了join类型的映射，示例如下（根据你的实际关系名称调整）：

{
  "mappings": {
    "properties": {
      "join_field": {
        "type": "join",
        "relations": {
          "parent": "child" // 这里"parent"是父文档类型，"child"是子文档类型
        }
      },
      // 其他业务字段...
    }
  }
}

2. 在KSQLDB中构造子数据格式 #

Elasticsearch要求子文档的join_field必须是包含name（子类型名）和parent（父文档ID）的嵌套结构，你可以用KSQL的STRUCT类型生成这个结构。

假设你的父表主键为parent_id，子表通过child_parent_id关联父表，子类型名为child，子数据的KSQL查询示例：

SELECT
  child_id AS id, // 子文档ID，可直接用子表主键或KSQL生成的ROWKEY
  STRUCT(
    name := 'child',
    parent := child_parent_id
  ) AS join_field,
  // 子表其他业务字段
  child_name,
  child_age
FROM child_table
EMIT CHANGES;

这里的核心是：

• name必须和ES映射中定义的子关系名称完全一致
• parent的值必须等于父文档在Elasticsearch中的_id（通常是父表的主键）

3. Elasticsearch Sink Connector关键配置 #

确保Connector配置支持嵌套结构，并正确处理文档ID，核心配置项如下：

connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=ksql_parent_topic,ksql_child_topic // 父、子数据对应的KSQL输出Topic
connection.url=http://elasticsearch:9200
type.name=_doc
key.ignore=false // 若用KSQL表主键作为ES文档ID，需开启此配置
schema.ignore=true // 若KSQL输出Schema无需严格匹配ES映射，可开启
transforms=unwrap
transforms.unwrap.type=io.confluent.connect.transforms.UnwrapFromEnvelope // 移除KSQL默认的消息包裹层

4. 验证父子关联 #

同步完成后，可通过ES的关联查询验证结果，比如查询属于某父文档的所有子文档：

GET /your_index/_search
{
  "query": {
    "has_parent": {
      "parent_type": "parent",
      "query": {
        "match": {
          "parent_name": "目标父文档名称"
        }
      }
    }
  }
}

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