JAVA高级面试进阶训练营视频教程

admin

 

 

　　今天用logstash同步数据库记录到elasticsearch时候出现错误，错误信息如下：

[2019-10-12T23:51:00,529][WARN ][logstash.inputs.jdbc     ] tracking_column not found in dataset. {:tracking_column=>"updated_time"}
{ 2018 rufus-scheduler intercepted an error:
  2018   job:
  2018     Rufus::Scheduler::CronJob "* * * * *" {}
  2018   error:
  2018     2018
  2018     TypeError
  2018     can't dup NilClass
  2018       org/jruby/RubyKernel.java:1882:in `dup'
  2018       uri:classloader:/META-INF/jruby.home/lib/ruby/stdlib/date/format.rb:838:in `_parse'
  2018       uri:classloader:/META-INF/jruby.home/lib/ruby/stdlib/date.rb:1830:in `parse'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-jdbc-4.3.13/lib/logstash/plugin_mixins/jdbc/value_tracking.rb:87:in `set_value'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-jdbc-4.3.13/lib/logstash/plugin_mixins/jdbc/jdbc.rb:237:in `execute_statement'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-jdbc-4.3.13/lib/logstash/inputs/jdbc.rb:277:in `execute_query'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/logstash-input-jdbc-4.3.13/lib/logstash/inputs/jdbc.rb:258:in `block in run'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:234:in `do_call'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:258:in `do_trigger'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:300:in `block in start_work_thread'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:299:in `block in start_work_thread'
  2018       org/jruby/RubyKernel.java:1292:in `loop'
  2018       /usr/local/logstash-6.4.3/vendor/bundle/jruby/2.3.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:289:in `block in start_work_thread'
  2018   tz:
  2018     ENV['TZ']: 
  2018     Time.now: 2019-10-12 23:51:00 +0800
  2018   scheduler:
  2018     object_id: 2008
  2018     opts:
  2018       {:max_work_threads=>1}
  2018       frequency: 0.3
  2018       scheduler_lock: #<Rufus::Scheduler::NullLock:0x5f5e76bc>
  2018       trigger_lock: #<Rufus::Scheduler::NullLock:0x707e50a2>
  2018     uptime: 242.19924699999999 (4m2s199)
  2018     down?: false
  2018     threads: 2
  2018       thread: #<Thread:0x76e69139>
  2018       thread_key: rufus_scheduler_2008
  2018       work_threads: 1
  2018         active: 1
  2018         vacant: 0
  2018         max_work_threads: 1
  2018       mutexes: {}
  2018     jobs: 1
  2018       at_jobs: 0
  2018       in_jobs: 0
  2018       every_jobs: 0
  2018       interval_jobs: 0
  2018       cron_jobs: 1
  2018     running_jobs: 1
  2018     work_queue: 0

　　

这种错误会导致创建不了时间节点文件。

 

问题原因是由于我的conf 中sql是多表联查的，我给每个字段起了别名（驼峰方式），但是logstash里面不支持驼峰字段，所以将字段别名全部用了小写，之后就好了，以下附上本人的conf文件

 

input {
  jdbc {
    jdbc_driver_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://XXXX/test"
    jdbc_user => "xxxx"
    jdbc_password => "xxxx"
    schedule => "* * * * *"
    statement => "SELECT p.product_id productid, p. NAME, p.title, p.description, pc.product_category_name productCategoryName, pc.product_category_describe productCategoryDescribe, ppc.price_category_name priceCategoryName, ppc.price_category_describe priceCategoryDescribe, pp.price, p.created_by createdBy, p.created_time createdTime, p.updated_by updatedBy, p.updated_time updatedtime FROM product p LEFT JOIN product_price pp ON p.product_id = pp.product_id LEFT JOIN product_category pc ON p.category_id = pc.product_category_id LEFT JOIN product_price_category ppc ON pp.price_category_id = ppc.price_category_id WHERE p.updated_time >= :sql_last_value"
    use_column_value => true
    tracking_column_type => "timestamp"
    tracking_column => "updatedtime"
    last_run_metadata_path => "syncpoint_product"
  }
}


output {
    elasticsearch {
        # ES的IP地址及端口
        hosts => ["172.31.70.140:9200","172.31.70.140:9201"]
        # 索引名称 可自定义
        index => "product"
        # 需要关联的数据库中有有一个id字段，对应类型中的id
        document_id => "%{productid}"
        document_type => "product"
    }
    stdout {
        # JSON格式输出
        codec => json_lines
    }
}

　　

 

本次整理希望能帮助大家。