Logstash聚合数据至Elasticsearch时章节数据丢失求助

我通过Logstash将数据推送至Elasticsearch，使用aggregate插件实现父子结构数据聚合，但出现无规律的数据丢失：部分卡片的章节能完整推送，有的卡片本该包含10个章节，却只推送3、5或7个，完全没规律，找不到问题所在，求排查方向和遗漏点。

Logstash配置文件 #

input { 
    jdbc {
        jdbc_connection_string => "my connection string"
        jdbc_user => "my user"
        jdbc_password => "my_password"
        jdbc_driver_library => "/usr/share/logstash/logstash-core/lib/jars/jdbc-mssql.jar"
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        last_run_metadata_path => "/etc/logstash/.logstash_jdbc_last_run_qa_flashcard"
        type => "card"
        statement => "select * from carddetailview where dateupdated > :sql_last_value"
        tracking_column => "dateupdated"
        tracking_column_type => "timestamp"
        use_column_value => "true"
        schedule => "* * * * *"
    }
}
filter{
    aggregate {
        task_id => "%{cardid}"
        code => "
        map['cardid']          = event.get('cardid')
        map['topicname']    = event.get('topicname')
        map['l1subject']    = event.get('l1subject')
        map['l2subject']    = event.get('l2subject')
        map['dateupdated']  = event.get('dateupdated')
        map['chapters'] ||= []
        map['chapters'] << 
                {   
                    'detailid' => event.get('detailid'),
                    'front'    => event.get('front'),
                    'back'    => event.get('back')
                }
        "
        push_previous_map_as_event => true
        timeout => 60
        timeout_tags => ['aggregated']
    }
    mutate { remove_field => ["detailid", "front", "back"] }        
}
output {
    elasticsearch {
      hosts => [ "http://localhost:9200" ]
          user => 'elastic'
          password => 'my_password'
          index => "myindex"
          document_type => "card"
          document_id => "%{cardid}" 
    }   
    stdout { codec => "json_lines" }
}

Elasticsearch示例数据（仅包含4个章节，实际应为21个） #

{
      "userid" : 20,
      "chapters" : [
        {
          "carddetailid" : 1246,
          "backscore" : 36,
          "front" : "Flamebait",
          "back" : "A"
        },
        {
          "carddetailid" : 1247,
          "backscore" : 42,
          "front" : "Meme",
          "back" : "B"
        },
        {
          "carddetailid" : 1248,
          "backscore" : 40,
          "front" : "Posts",
          "back" : "C"
        },
        {
          "carddetailid" : 1249,
          "backscore" : 38,
          "front" : "Chats",
          "back" : "D"
        }
      ],
      "keyword" : "A program that appears desirable",
      "cardid" : "1"
    }

排查方向与解决方案 #

1. 修正aggregate插件的推送逻辑 #

当前配置中push_previous_map_as_event => true会导致每收到一个同cardid的新事件，就推送一次之前的不完整聚合结果，后续事件会覆盖ES中的文档，但如果中途触发超时，最终保留的就是不完整的章节数。建议调整为：

aggregate {
    task_id => "%{cardid}"
    code => "
    # 原code逻辑不变
    "
    push_previous_map_as_event => false  # 关闭提前推送
    push_map_as_event_on_timeout => true # 仅超时或所有事件处理完时推送完整聚合结果
    timeout => 300  # 适当调大超时时间，适配数据拉取延迟
    timeout_tags => ['aggregated']
}

2. 确保同一卡片的章节事件连续处理 #

aggregate插件的map是线程隔离的，若同一cardid的章节事件被分到不同Logstash worker线程，每个线程会维护独立的map，最终推送不完整数据：

• 临时验证：启动Logstash时指定单worker：bin/logstash -f your_config.conf --pipeline.workers 1，观察是否还会出现章节丢失。
• 长期优化：修改SQL查询，确保同一卡片的所有章节连续输出：

  select * from carddetailview where dateupdated > :sql_last_value order by cardid, detailid
  

  同时设置jdbc_fetch_size控制每次拉取行数，避免数据分批过于零散：

  jdbc {
      # 其他配置不变
      jdbc_fetch_size => 1000
  }
  

3. 修正数据拉取的完整性逻辑 #

当前SQL仅拉取dateupdated大于上次运行时间的章节，若卡片下部分章节未更新，会导致遗漏：

• 修改SQL，确保只要卡片有更新（或任意章节更新），拉取该卡片的所有章节：

  select * from carddetailview 
  where cardid in (
      select distinct cardid from carddetailview where dateupdated > :sql_last_value
  )
  

• 检查last_run_metadata_path文件，确认dateupdated的跟踪值是否正确，避免因时间精度问题漏掉数据。

4. 开启debug日志追踪聚合过程 #

开启Logstash debug日志，查看aggregate插件的事件处理细节：

• 在logstash.yml中添加：logging.level: debug，或启动时加--log.level debug
• 过滤日志中的aggregate关键词，跟踪同一cardid的事件处理流程，确认哪些章节事件被加入map，以及超时触发的时间点，定位是否有事件未被正确处理。

5. 检查字段名一致性 #

示例数据中章节字段为carddetailid，但配置中code部分使用detailid，存在字段名不匹配风险，需确保SQL返回的字段名与event.get()调用的字段完全一致，避免因字段缺失导致章节未被加入聚合列表。

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