使用Logstash将Elasticsearch数据导入MySQL时遇配置错误求助

错误根源 #

你的配置存在两个核心问题，直接导致了Logstash的配置解析错误：

1. parameters参数格式错误：Logstash的JDBC Output插件中，parameters需要是键值对哈希结构，而非数组，用来映射SQL占位符与Elasticsearch字段值。
2. 冗余无效配置：schedule是JDBC Input插件的定时拉取参数，放在JDBC Output中完全无效，会干扰配置解析逻辑。

修正后的配置文件 #

elasticsearch {
    hosts => "localhost:9200"
    index => "testrak"
    query => '{"query": { "match_all": {} }}'
    scroll => "5m"
    docinfo => true # 若需获取Elasticsearch文档的_id，必须开启此选项
  }
}

output {
  jdbc {
    jdbc_driver_library => "C:/Program Files (x86)/MySQL/Connector J 8.0/mysql-connector-java-8.0.29.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/elasticdata"
    jdbc_user => "root"
    jdbc_password => ""
    # 方式1：直接在SQL中用%{field}引用Elasticsearch字段（推荐简单场景使用）
    statement => "INSERT INTO elastic_table (id, incident_number, created_on, service_name, ed) VALUES ('%{[@metadata][_id]}', '%{incident_number}', '%{created_on}', '%{service_name}', '%{ed}')"
    
    # 方式2：使用parameters哈希映射占位符（适合复杂参数场景）
    # statement => "INSERT INTO elastic_table (id, incident_number, created_on, service_name, ed) VALUES (:doc_id, :incident_num, :create_time, :service, :ed_val)"
    # parameters => {
    #   "doc_id" => "%{[@metadata][_id]}",
    #   "incident_num" => "%{incident_number}",
    #   "create_time" => "%{created_on}",
    #   "service" => "%{service_name}",
    #   "ed_val" => "%{ed}"
    # }
  }
}

额外注意事项 #

• 若你要插入的id是Elasticsearch文档的原生_id，必须开启docinfo => true，并通过%{[@metadata][_id]}引用该字段。
• 确保MySQL表elastic_table的字段类型与Elasticsearch对应字段完全匹配（比如日期字段格式、字符串长度等），避免插入失败。
• 提前用logstash -f 你的配置文件名.conf --config.test_and_exit命令测试配置语法，快速定位错误。
• 执行迁移时可添加--config.reload.automatic参数，修改配置后无需重启Logstash即可自动重载。

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