如何配置Logstash JDBC输入用WHERE替代OFFSET查询PostgreSQL

解决方案 #

可以通过自定义分页逻辑替代Logstash默认的LIMIT+OFFSET分页，核心是利用跟踪字段记录每页最后一条数据的标记值，通过WHERE条件过滤已同步数据，避免全表扫描。具体调整如下：

1. 调整Logstash配置 #

关闭默认分页，自定义SQL中的分页条件，同时配置跟踪字段记录每页的结束标记：

input {
  file {
    path => "/var/log/logstash/logstash-plain.log"
    type => "logstash-logs"
    start_position => "beginning"
  }

  jdbc {
    jdbc_driver_library => "/usr/share/logstash/external_jars/postgresql-42.5.4.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://172.17.0.1:5432/tyver_stage"
    jdbc_user => "tyver_stage"
    jdbc_password => "password"
    schedule => "*/5 * * * *"
    # 自定义SQL，加入基于order_column的分页过滤条件
    statement => "
      SELECT
        c.*,
        COALESCE(c.updated_at, c.created_at) AS order_column,
        CASE
          WHEN ARRAY_AGG(ucv.user_id) = ARRAY[null]::integer[] THEN ARRAY[]::integer[]
          ELSE ARRAY_AGG(ucv.user_id)
        END AS viewed_by
      FROM
        creatives as c
          LEFT JOIN
        user_creative_views ucv ON ucv.creative_id = c.id
      WHERE
        -- 增量同步条件：保留原有的时间过滤逻辑
        (c.updated_at >= :sql_last_value OR (c.updated_at IS NULL AND c.created_at >= :sql_last_value))
        -- 分页条件：只取order_column大于上一页最后一条记录的值
        AND COALESCE(c.updated_at, c.created_at) > :last_page_last_value
      GROUP BY
        c.id
      ORDER BY
        COALESCE(c.updated_at, c.created_at) ASC
      -- 限制每页数据量，替代OFFSET
      LIMIT 10000
    "
    use_column_value => true
    tracking_column => "order_column"
    tracking_column_type => "timestamp"
    # 禁用默认分页逻辑
    jdbc_paging_enabled => false
    record_last_run => true
    clean_run => false
    # 初始化分页起始值，确保第一次能查询所有符合增量条件的数据
    parameters => { "last_page_last_value" => "1970-01-01 00:00:00.000000+0000" }
    # 存储每页最后一条记录的order_column值，作为下一页的起始条件
    last_run_metadata_path => "/usr/share/logstash/.last_run"
  }
}

filter {
  # 可根据需求添加数据处理逻辑
}

output {
  elasticsearch {
    hosts => ["172.17.0.1:9200"]
    index => "tyver_index_creatives"
    document_id => "%{id}"
  }
}

2. 数据库索引优化 #

为了让WHERE中的COALESCE(c.updated_at, c.created_at)条件高效执行，必须创建对应的函数索引：

CREATE INDEX idx_creatives_order_column ON creatives (COALESCE(updated_at, created_at));

3. 处理重复标记值的情况 #

如果多条记录的order_column值完全相同，可能导致数据漏同步或重复同步。此时可以结合主键id来确保分页的唯一性，修改SQL如下：

SELECT
  c.*,
  COALESCE(c.updated_at, c.created_at) AS order_column,
  c.id AS last_page_last_id,
  CASE
    WHEN ARRAY_AGG(ucv.user_id) = ARRAY[null]::integer[] THEN ARRAY[]::integer[]
    ELSE ARRAY_AGG(ucv.user_id)
  END AS viewed_by
FROM
  creatives as c
    LEFT JOIN
  user_creative_views ucv ON ucv.creative_id = c.id
WHERE
  (c.updated_at >= :sql_last_value OR (c.updated_at IS NULL AND c.created_at >= :sql_last_value))
  AND (
    COALESCE(c.updated_at, c.created_at) > :last_page_last_value
    OR (COALESCE(c.updated_at, c.created_at) = :last_page_last_value AND c.id > :last_page_last_id)
  )
GROUP BY
  c.id
ORDER BY
  COALESCE(c.updated_at, c.created_at) ASC, c.id ASC
LIMIT 10000

同时在Logstash配置中新增last_page_last_id参数，跟踪主键值，确保分页逻辑的准确性。

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