如何通过Logstash为ElasticSearch中Staff文档批量添加Department_Id字段

根据你的场景，提供两种易上手的方案，可根据业务情况选择：

方案一：Logstash全量重新同步（新手友好） #

利用你已熟悉的Logstash同步流程，直接从MySQL关联查询完整数据，覆盖ES中现有Staff文档，一次性补全Department_Id字段。

步骤1：编写Logstash配置文件（示例命名为staff_full_sync.conf） #

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://你的MySQL地址:3306/数据库名?useSSL=false&serverTimezone=UTC"
    jdbc_user => "MySQL用户名"
    jdbc_password => "MySQL密码"
    jdbc_driver_library => "/path/to/mysql-connector-java-8.0.XX.jar" # 替换为你的驱动路径
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    # 关联查询直接获取带Department_Id的完整Staff数据
    statement => "
      SELECT 
        s.Id AS Staff_Id, 
        s.Name AS Staff_Name, 
        d.Name AS Department_Name, 
        d.Id AS Department_Id 
      FROM STAFF s
      LEFT JOIN DEPARTMENT d ON s.Department_Id = d.Id
    "
  }
}

output {
  elasticsearch {
    hosts => ["http://你的ES地址:9200"]
    index => "staff" # 替换为你的Staff文档所在ES索引名
    document_id => "%{Staff_Id}" # 用Staff_Id作为文档ID，避免重复创建，直接覆盖旧文档
  }
}

步骤2：启动Logstash执行同步 #

在Logstash安装目录执行命令：

bin/logstash -f staff_full_sync.conf

等待同步完成后，ES中的Staff文档就会自动带上Department_Id字段。

方案二：ES Update By Query批量更新（无需全量同步） #

如果不想重新同步1000万条数据，可先将部门名称-ID映射导入ES临时索引，再批量更新现有Staff文档。

步骤1：同步部门映射到ES临时索引 #

编写Logstash配置文件dept_map_sync.conf：

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://你的MySQL地址:3306/数据库名?useSSL=false&serverTimezone=UTC"
    jdbc_user => "MySQL用户名"
    jdbc_password => "MySQL密码"
    jdbc_driver_library => "/path/to/mysql-connector-java-8.0.XX.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    statement => "SELECT Id AS Department_Id, Name AS Department_Name FROM DEPARTMENT"
  }
}

output {
  elasticsearch {
    hosts => ["http://你的ES地址:9200"]
    index => "department_map" # 临时存储部门映射的索引名
    document_id => "%{Department_Id}"
  }
}

启动同步：

bin/logstash -f dept_map_sync.conf

步骤2：批量更新Staff索引 #

通过ES的_update_by_query API，基于部门名称匹配临时索引中的ID，补全字段：

curl -X POST "http://你的ES地址:9200/staff/_update_by_query?conflicts=proceed" -H 'Content-Type: application/json' -d'
{
  "script": {
    "source": "ctx._source.Department_Id = params.deptMap.get(ctx._source.Department_Name);",
    "params": {
      "deptMap": {}
    },
    "lang": "painless"
  },
  "query": {
    "bool": {
      "must_not": {
        "exists": {
          "field": "Department_Id"
        }
      }
    }
  },
  "size": 1000
}
'

注：如果部门数量过多（10万条），直接传递映射参数可能触发大小限制，可先将department_map索引的数据导出为JSON对象，分批次传入脚本参数，或使用ES的terms lookup优化查询逻辑。

注意事项 #

• 方案一操作简单，适合新手，但全量同步1000万条数据耗时较长，建议在业务低峰期执行。
• 方案二更高效，但需要对ES基础API有初步了解，执行前请先在测试环境验证。
• 无论选择哪种方案，执行前请备份ES的Staff索引，避免数据异常。

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