如何优化Logstash从PostgreSQL批量导入1.4亿条数据的性能

• 导入耗时过长：初始导入1.4亿条数据需48小时，当前仍需约17小时；
• 分页查询效率衰减：首页查询耗时8秒，到第4000万条数据时每页耗时120秒，最终甚至接近15分钟。

• 将pipeline.workers数量调整至与机器核心数一致，性能提升显著，但当前CPU使用率已偏高，不确定是否应继续增加；
• 逐步将jdbc_page_size从10万提升至25万，性能持续改善，但更大数值会导致系统崩溃；
• 调整pipeline.batch_size至5000，无明显性能变化；
• 因无法动态设置分页键，无法使用keyset分页；
• 机器已耗尽可用内存。

输入配置 #

input {
  jdbc {
    jdbc_driver_library => "/usr/share/logstash/postgresql-42.3.1.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://@host:$port/MYDB"
    jdbc_user => "user"
    jdbc_password => "password"
    #schedule => "* * * * *"
    statement => "SELECT date_trunc('hour', client_received_timestamp) as client_received_timestamp, is_error, httpsStatusCode from MYSCHEMA.MY_TABLE WHERE client_received_timestamp >= '2023-05-01 00:0:0' AND client_received_timestamp < '2023-06-01 00:0:0' UNION ALL SELECT date_trunc('hour', client_received_timestamp) as client_received_timestamp, is_error, httpsStatusCode from MYSCHEMA.MY_SECOND_TABLE WHERE client_received_timestamp >= '2023-05-01 00:0:0' AND client_received_timestamp < '2023-06-01 00:0:0' UNION ALL SELECT date_trunc('hour', client_received_timestamp) as client_received_timestamp, is_error, httpsStatusCode from MYSCHEMA.MY_THIRD_TABLE WHERE client_received_timestamp >= '2023-05-01 00:0:0' AND client_received_timestamp < '2023-06-01 00:0:0'"
    tracking_column => client_received_timestamp
    clean_run => false
    last_run_metadata_path => "/usr/share/logstash/queues/last_run_update_time"
    jdbc_page_size => 250000
    jdbc_paging_enabled => true
  }
}

filter {
  mutate {
    add_field => { "SCHEMA" => "MYSCHEMA" }
    add_field => { "TABLE" => "MY_TABLE" }
  }
}

output {
    elasticsearch {
        hosts       => ["https://elasticsearch:9200"]
        user        => "user"
        password    => "pass"
        index       => "logs-from-db-v1-%{+YYYY.MM.dd}"
    }
}

Pipeline配置 #

- pipeline.id: my-db-pipeline
  path.config: "/etc/path/to/db-v1.config"
  pipeline.workers: 8
  pipeline.batch.size: 5000

数据库端优化 #

• 拆分查询并行拉取：把原SQL的UNION ALL拆分为3个独立的JDBC输入插件，每个插件对应一张表；同时将时间范围按天拆分，每个输入处理一个小时间窗口的数据集，利用多输入并行能力降低单查询负载。
• 预计算聚合结果：在PostgreSQL中创建物化视图，按小时预聚合client_received_timestamp、is_error、httpsStatusCode的结果，直接从物化视图拉取数据，减少查询时的实时计算开销。
• 优化索引：确保三张表的client_received_timestamp字段有B-tree索引，同时创建包含is_error、httpsStatusCode的覆盖索引，让查询无需回表，提升分页性能。
• 临时调整数据库参数：调大work_mem至64MB，减少分页排序时的磁盘临时文件使用；调大shared_buffers至机器内存的25%，增强数据库缓存能力。

Logstash端优化 #

• 启用磁盘队列：在Pipeline配置中添加queue.type: persisted，指定path.queue存储路径，将数据从内存队列转移到磁盘队列，缓解内存耗尽问题；同时调整queue.max_bytes和queue.page_capacity适配磁盘空间。
• 优化Elasticsearch输出：
  • 设置bulk_max_size为250000（与jdbc_page_size匹配），减少批量请求次数；
  • 添加flush_interval => 1，避免频繁小批量写入；
  • 开启parallel_bulk => true，设置parallel_bulk_threads => 3，并行发送批量请求（需Elasticsearch集群有足够处理能力）。
• 清理Filter逻辑：当前Filter给所有数据添加固定TABLE字段，与实际三张表的UNION逻辑冲突。若业务不需要SCHEMA、TABLE字段，直接删除Filter；若需要，在每个JDBC输入的独立Filter中设置对应表名，避免无效字段操作。
• 调整JVM参数：设置LS_JAVA_OPTS="-Xms16g -Xmx16g -XX:+UseG1GC"，给Logstash分配足够堆内存（不超过机器可用内存的50%），同时用G1GC优化垃圾回收效率。

架构层面优化 #

• 分布式拉取：部署多个Logstash实例，每个实例处理不同的时间分片或表数据，统一写入Elasticsearch集群，提升整体吞吐量。
• 引入中间缓冲层：用Kafka作为PostgreSQL和Logstash之间的缓冲，先将数据批量导出到Kafka，再用多个Logstash消费者从Kafka读取数据写入Elasticsearch，实现生产消费解耦，降低数据库压力。

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