使用Logstash向ELK导入大CSV文件时数据导入不全求助

针对每日800M/350万行级CSV文件导入ES时部分日期数据不全、无报错的问题，结合你的配置给出以下排查和修复方案：

1. 清理Sincedb残留读取记录 #

你的配置中使用了sincedb_path => "csv_data.db"，Logstash会通过该文件记录每个文件的读取位置。若之前未处理完的文件记录残留，或后续同命名规则的文件inode未变化，Logstash会从上次中断位置开始读取，导致数据不全。

• 临时解决：每次导入前手动删除csv_data.db文件，强制Logstash从头读取文件。
• 永久配置：将sincedb_path设置为/dev/null（Linux环境），彻底禁用sincedb记录，确保每次都从start_position => "beginning"读取：

  input {
    file {
          path => "/home/importdata/pack_*.csv"
          start_position => "beginning"
          sincedb_path => "/dev/null"
    }
  }
  

2. 排查CSV格式解析错误 #

部分日期的CSV可能存在格式异常（如字段内包含未转义的分隔符;、字段内换行、缺少引号包裹等），Logstash的csv插件默认不会抛出致命错误，仅会给错误数据添加_csvparsefailure标签，日志无报错但数据会丢失。

• 捕获错误数据：在output中添加错误输出，将解析失败的行导出到文件排查：

  output {
    if "_csvparsefailure" in [tags] {
      file {
        path => "/home/importdata/csv_parse_errors_%{+YYYYMMDD}.log"
        codec => line { format => "%{message}" }
      }
    }
    elasticsearch {
      hosts => ["http://ip1:9200","http://ip2:9200","http://ip3:9200"]
      index => "%{filename}"
    }
  }
  

• 优化CSV解析配置：添加quote_char参数适配带引号的字段，避免解析错误：

  csv {
          separator => ";"
          skip_header => "true"
          quote_char => "\""
          columns => ["id","cif","global_id","cus_name","cus_dob","cus_address","cus_email","cus_phone","cus_branch","cus_acct_exec_code","cus_acct_exec_name","cus_branch_name","created_by","created_time","updated_by","updated_time","is_deleted","status","deleted_by","deleted_time","route","client_type","client_group"]              
      }
  

3. 提升Logstash处理性能 #

大文件导入时，Logstash默认的处理线程和批处理配置可能存在性能瓶颈，导致数据堆积未完全导入。

• 调整Pipeline参数：在logstash.yml中修改以下配置，提升处理能力：

  pipeline.workers: 8  # 根据CPU核心数调整，建议为核心数的1-2倍
  pipeline.batch.size: 1000  # 每次批处理的事件数
  pipeline.batch.delay: 50  # 批处理延迟时间（毫秒）
  

• 确保文件稳定：导入过程中不要修改、删除或移动CSV文件，直到Logstash日志中出现该文件的closed标记，确认处理完成。

4. 检查ElasticSearch写入限制 #

ES集群可能因磁盘水位线、分片异常等原因静默拒绝写入，无明显报错但数据丢失。

• 验证索引文档数：执行ES命令对比CSV行数（排除表头）与索引文档数：

  # 统计CSV行数（排除表头）
  wc -l /home/importdata/pack_xxxxxx.csv | awk '{print $1-1}'
  # 统计ES索引文档数
  curl -XGET http://ip1:9200/pack_xxxxxx/_count?pretty
  

• 添加唯一ID校验：在output的elasticsearch配置中指定document_id为CSV的唯一字段（如id），并设置action => "create"，若存在重复ID会直接报错，便于排查：

  elasticsearch {
    hosts => ["http://ip1:9200","http://ip2:9200","http://ip3:9200"]
    index => "%{filename}"
    document_id => "%{id}"
    action => "create"
  }
  

• 检查ES集群状态：查看磁盘使用率（确保未达85%的low watermark）、分片健康状态：

  curl -XGET http://ip1:9200/_cat/disk?pretty
  curl -XGET http://ip1:9200/_cat/shards/pack_xxxxxx?pretty
  

5. 确认文件权限与唯一性 #

• 确保Logstash进程对/home/importdata/目录及CSV文件有读权限：

  chown -R logstash:logstash /home/importdata/
  

• 确保每日CSV文件名唯一（如添加日期后缀pack_20240520.csv），避免因文件名重复导致的sincedb识别异常。

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