无唯一ID/时间戳时双节点Logstash JDBC重复拉取700万数据求助

一、解决持续拉取不停止的问题 #

核心原因 #

• 查询缺少稳定的ORDER BY子句，导致分页结果无序且不稳定，Logstash反复拉取重复数据，无法终止同步
• 未明确配置一次性执行逻辑，默认触发重复查询

配置修改 #

更新JDBC输入配置，添加分页控制和一次性执行参数：

input {
  jdbc {
    id => "database"
    jdbc_driver_library => "my_driver"
    jdbc_driver_class => "my_driver_class"
    jdbc_connection_string => "my_connection"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_paging_enabled => true
    jdbc_page_size => 10000 # 按数据库性能调整，推荐5000-20000
    schedule => "" # 空字符串表示仅执行一次全量同步
    statement_filepath => "mysql_query.sql"
  }
}

同时修改mysql_query.sql，添加稳定排序规则：

-- 选择表中1-多个字段组合，确保排序结果固定
SELECT * FROM your_table
ORDER BY column1, column2, column3;

如果无可用排序字段，可临时用MySQL内部标识保证排序稳定：

SELECT * FROM your_table
ORDER BY @@server_id, table_name; -- 利用服务器ID固定排序基准

二、双节点分片处理实现并行同步 #

由于数据库无唯一ID/时间戳，通过行号分片让两个节点各自处理一半数据，避免重复，提升同步效率。

1. 为节点分配唯一标识 #

启动两个节点时分别传入环境变量：

• 节点1：NODE_ID=1 ./bin/logstash -f mypipeline.conf
• 节点2：NODE_ID=0 ./bin/logstash -f mypipeline.conf

2. 修改SQL实现分片查询 #

MySQL 8+版本（支持CTE窗口函数）：

更新mysql_query.sql为：

WITH numbered_rows AS (
    SELECT *, ROW_NUMBER() OVER (ORDER BY column1, column2) AS row_num
    FROM your_table
)
SELECT * FROM numbered_rows
WHERE MOD(row_num, 2) = ${NODE_ID};

MySQL 5.x版本（用变量生成行号）：

SELECT * FROM (
    SELECT *, @row := @row + 1 AS row_num
    FROM your_table, (SELECT @row := 0) AS init
    ORDER BY column1, column2
) AS numbered_rows
WHERE MOD(row_num, 2) = ${NODE_ID};

3. 配置Logstash解析环境变量 #

将JDBC输入的statement_filepath替换为statement，显式传入参数：

input {
  jdbc {
    # 其他参数不变
    statement => "WITH numbered_rows AS (SELECT *, ROW_NUMBER() OVER (ORDER BY column1, column2) AS row_num FROM your_table) SELECT * FROM numbered_rows WHERE MOD(row_num, 2) = ?"
    parameters => { "node_id" => "${NODE_ID}" }
  }
}

三、额外优化建议 #

• 避免Elasticsearch重复索引：在输出配置中添加document_id，用多个字段组合生成唯一ID：

  output {
    elasticsearch {
      # 其他参数不变
      document_id => "%{column1}_%{column2}_%{column3}"
    }
  }
  

• 监控处理进度：通过Logstash监控API（/_node/stats/pipelines）查看两个节点的事件处理量，确保进度正常
• 调整资源配置：根据服务器性能，修改jvm.options中的堆内存大小（比如设置为-Xms4g -Xmx4g），避免内存溢出

内容的提问来源于stack exchange，提问作者Murat K.