Logstash同步MySQL至ES：日期格式转换失败求助

你遇到的核心问题是日期字段的源格式和你在date过滤器中指定的匹配模式不匹配，导致了解析失败。MySQL的datetime/timestamp类型字段会被JDBC驱动自动转换成ISO 8601格式的字符串（比如2020-02-28T09:18:18.000Z），而你原来的date过滤器用了yyyy-MM-dd HH:mm:ss去匹配，自然会触发_dateparsefailure。

下面给你两种可行的解决方案，你可以根据需求选择：

方案一：先解析日期对象，再格式化字符串（推荐，时区更可控） #

这种方法先把ISO格式的字符串解析成Logstash的日期对象，再将其格式化为你需要的yyyy-MM-dd HH:mm:ss格式，同时可以处理时区问题：

input {
  jdbc { 
    jdbc_driver_library => "/etc/mysql/driver/mysql-connector-java-5.1.48/mysql-connector-java-5.1.48-bin.jar" 
    jdbc_driver_class => "com.mysql.jdbc.Driver" 
    jdbc_connection_string => "jdbc:mysql://localhost:3306/db" 
    jdbc_user => root 
    jdbc_password => "secret" 
    tracking_column => "id" 
    use_column_value=> true 
    statement => "SELECT * FROM db.logs;" 
    schedule => "* * * * * *" 
    # 可选：设置JDBC时区，确保和MySQL服务器时区一致，避免时间偏移
    jdbc_default_timezone => "UTC"
  } 
} 

filter {
  # 第一步：解析ISO格式的日期字符串为Logstash日期对象
  date {
    match => ["date", "ISO8601"] # 匹配ISO 8601格式
    target => "internal_date" # 将解析后的日期对象存入临时字段
    timezone => "UTC" # 根据你的实际时区调整，比如"Asia/Shanghai"
  }

  # 第二步：将日期对象格式化为目标字符串格式
  ruby {
    code => "event.set('formatted_date', event.get('internal_date').strftime('%Y-%m-%d %H:%M:%S'))"
  }

  # 第三步：清理临时字段，并重命名格式化后的字段为原字段名（可选）
  mutate {
    remove_field => ["date", "internal_date"]
    rename => {"formatted_date" => "date"}
  }
}

output { 
  elasticsearch { 
    document_id=> "%{id}" 
    document_type => "_doc" 
    index => "logs" 
    hosts => "http://localhost:9200" 
    sniffing => true 
  } 
  stdout{ codec => rubydebug } 
}

方案二：直接字符串替换（简单快速，适合不需要时区转换的场景） #

如果你的日期不需要处理时区，只是单纯去掉T和.000Z，可以直接用mutate的gsub来替换字符串：

filter {
  mutate {
    gsub => [
      "date", "T", " ",  # 替换T为空格
      "date", "\.\d+Z$", "" # 去掉末尾的毫秒和Z（比如.000Z）
    ]
  }

  # 可选：如果需要将字符串转为ES的日期类型，添加这一步
  date {
    match => ["date", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
  }
}

关键注意点 #

1. 时区一致性：确保MySQL服务器、Logstash的jdbc_default_timezone、date过滤器的timezone三者时区一致，避免出现时间偏移问题。
2. 字段名确认：确保你要处理的日期字段名确实是date，如果MySQL表中的字段名不同，要对应修改配置中的字段名。
3. 测试验证：修改配置后，先通过stdout的rubydebug codec查看处理后的字段格式，确认正确后再同步到Elasticsearch。

内容的提问来源于stack exchange，提问作者M. Badovsky