Logstash JDBC插件无限循环写入ES 数据量超数据库实际行数

• 调度规则配置错误：schedule => "*/5 * * * * *"为6位秒级cron表达式，实际触发频率为每5秒执行一次全表查询，远高于常规同步频率，任务持续堆积会触发进程异常循环，必须手动杀进程才能终止。
• Elasticsearch输出段缺失唯一文档ID配置：未指定document_id参数时，Logstash会为每一条拉取到的记录生成随机UUID作为ES文档ID，同一条数据库记录每次同步都会作为新文档写入，直接导致索引内文档量持续翻倍增长，永远和数据库表行数无法对齐。
• 同步逻辑无增量跟踪：当前执行的SQL为全表无差别查询，没有记录上次同步的位置，每次调度都会拉取整表数据，配合高频调度不仅会给数据库造成不必要的查询压力，还会进一步放大重复写入的问题。

第一步：清理历史脏数据 #

先清空目标索引内的重复历史数据，避免旧数据干扰计数结果：

curl -XDELETE -u test:test http://localhost:9200/my_index

第二步：修正Logstash配置 #

核心调整三个模块：

• 将cron调度改为标准5位格式，按实际业务需求设置同步间隔，示例为每5分钟同步一次
• ES输出段配置document_id映射数据库表的主键字段，保证同一条数据库记录写入时自动覆盖ES中已存在的同ID文档，从根源杜绝重复文档生成
• 开启JDBC增量同步跟踪，持久化记录每次同步的截止位置，后续调度只拉取新增/变更的数据，避免每次全表扫描引发的进程异常

修正后的完整配置如下：

input {
  jdbc {
    jdbc_driver_library => "/correct_path/java/mysql-connector-java-8.0.27.jar" 
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/my_db"
    jdbc_user => "user" 
    jdbc_password => "password" 
    jdbc_paging_enabled => true
    jdbc_page_size => 1000
    # 修正为5位标准cron，示例为每5分钟执行一次
    schedule => "*/5 * * * *"
    # 开启增量列值跟踪
    use_column_value => true
    # 按实际表结构替换为主键字段，若用更新时间同步则替换为update_time字段
    tracking_column => "id"
    tracking_column_type => "numeric"
    # 持久化存储上次同步的位置，避免进程重启后从头拉取
    last_run_metadata_path => "/var/lib/logstash/jdbc_meta/my_table_last_run"
    # 增量查询SQL，只拉取上次同步位点之后的数据
    statement => "select * from my_table where id > :sql_last_value order by id asc"
  }
}

output {
    elasticsearch {
      user => "test"
      password => "test"
      hosts => ["localhost:9200"] 
      index => "my_index"
      # 绑定数据库主键为ES文档ID，同ID自动覆盖
      document_id => "%{id}"
    }
    stdout { codec => "rubydebug" }
}

注意：如果业务表使用更新时间字段做增量跟踪，将tracking_column改为表中存储更新时间的字段名，tracking_column_type设置为timestamp，对应SQL调整为select * from my_table where update_time > :sql_last_value order by update_time asc即可，document_id必须映射表中唯一非空的业务主键，不能使用随机生成值。

第三步：结果校验 #

启动Logstash等待首次全量同步完成后，分别查询两边的数据量做比对：

• 数据库端执行select count(*) from my_table;获取表实际行数
• ES端执行以下命令获取索引文档总数：

curl -XGET -u test:test http://localhost:9200/my_index/_count

两个返回值完全一致即符合要求，后续增量同步过程中因为有唯一文档ID做覆盖保障，不会再出现文档数虚高的问题。

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