elasticsearch - 问题从输入中的logstash连接到 Elasticsearch

Question

我想从oracle导入数据，并希望将导入数据的参数之一传递给 Elasticsearch 以获取其他一些细节。

例如:-如果我有一个从oracle db获得的Employee ID，例如说100行，我想将所有这100个雇员ID传递给 Elasticsearch ，并获得emp名称和薪水。

我现在能够从oracle检索数据，但无法连接到 Elasticsearch 。另外，我不确定哪种方法更好。
我正在使用日志存储库2.3.3和 Elasticsearch 日志存储库过滤器插件。

 input {
       jdbc {
        jdbc_connection_string => "jdbc:oracle:thin:@<dbhost>:<port>:<sid>"
        # The user we wish to execute our statement as
        jdbc_user => “user"
        jdbc_password => “pass"
        # The path to our downloaded jdbc driver
        jdbc_driver_library => “<path>"
        # The name of the driver class for oracle
        jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
        # our query
        statement => "SELECT empId, desg from Employee"
    }
    elasticsearch {
      hosts => "https://xx.corp.com:9200"
        index => “empdetails”
    }
   }
   output {
    stdout { codec => json_lines }
   }

由于 Elasticsearch ，出现以下错误。

插件有不可恢复的错误。将重新启动此插件。

插件:[“ https://xx.corp.com:9200”]，索引=>“empdetails”，查询=>“empId:'1001'“，编解码器=>” UTF-8“>，扫描=> true，大小=> 1000，滚动=>” 1m“，docinfo => false，docinfo_target =>” @@ metadata“，docinfo_fields => [” _ index“，” _ type“，” _ id“]，ssl => false>

错误:[401] {:level =>:error}

Answer 1

您需要使用 elasticsearch filter而不是 elasticsearch input

input {
   jdbc {
    jdbc_connection_string => "jdbc:oracle:thin:@<dbhost>:<port>:<sid>"
    # The user we wish to execute our statement as
    jdbc_user => “user"
    jdbc_password => “pass"
    # The path to our downloaded jdbc driver
    jdbc_driver_library => “<path>"
    # The name of the driver class for oracle
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    # our query
    statement => "SELECT empId, desg from Employee"
   }
}
filter {
  elasticsearch {
    hosts => ["xx.corp.com:9200"]
    query => "empId:%{empId}"
    user => "admin"
    password => "admin"
    sort => "empName:desc"
    fields => {
      "empName" => "empName" 
      "salary" => "salary" 
    }
  }
}
output {
  stdout { codec => json_lines }
}

结果，通过JDBC提取的每个记录将被ES中的相应数据所丰富。