Logstash同步PostgreSQL至Elasticsearch出现数据重复及索引重建问题求助

一、重复数据问题修复 #

你当前Elasticsearch输出配置中document_id参数被注释，且取值配置错误：

• 你从PostgreSQL查询出的表主键字段为id，直接用该字段作为ES的文档ID即可保证同一条数据更新时覆盖旧文档，不会生成重复数据。
• 将output的elasticsearch配置中的document_id行修改为：

document_id => "%{id}"

你已开启doc_as_upsert => true，该配置配合指定document_id即可实现数据存在则更新、不存在则插入，完全避免重复。

二、全量重建索引（增量不生效）问题修复 #

该问题核心原因是sql_last_value未被正确更新，导致每次定时任务都拉取全表数据，有3个需修正的配置点：

1. 关闭字段名自动小写转换
   PostgreSQL中你使用驼峰命名的"updatedAt"字段，Logstash jdbc插件默认会将所有查询字段名转为小写，导致无法识别tracking_column配置的updatedAt，无法更新sql_last_value，每次都用初始时间（1970-01-01）作为查询条件，拉取全表数据。
   在jdbc输入配置中添加参数：

lowercase_column_names => false

2. SQL查询添加排序规则
   sql_last_value默认取值为本次查询结果集最后一行的tracking_column值，你当前的SQL未按updatedAt排序，可能导致最后一行的updatedAt不是本次查询的最大值，sql_last_value更新错误，下次查询仍会拉取到旧数据。
   修改你的SQL语句，末尾添加排序：

SELECT id, title, description, type,"updatedAt" FROM public.videos WHERE ("updatedAt" > :sql_last_value AND "updatedAt" < NOW()) ORDER BY "updatedAt" ASC

3. （容器部署场景）持久化运行时元数据
   如果你用Docker/K8s部署Logstash，默认没有挂载last_run_metadata_path对应的文件，重启Logstash后会丢失之前记录的sql_last_value，重新拉取全量数据。你可以在jdbc配置中显式指定该路径，并挂载对应目录：

last_run_metadata_path => "/usr/share/logstash/config/last_run/video"

可选检查项 #

如果修改后仍存在增量不生效的问题，检查PG的updatedAt字段时区和Logstash运行时区是否一致，可以在jdbc连接串上加serverTimezone=Asia/Shanghai（替换为你所在的时区）避免时间差导致的sql_last_value计算错误。

修改后重启Logstash即可生效，首次启动会拉取全量数据，后续定时任务只会拉取updatedAt大于上次同步时间的增量数据。

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