logstash配置檔案如下：

input {   jdbc {     # mysql相關jdbc配置     jdbc_connection_string => "jdbc:mysql://localhost:3306/automall_data?useUnicode=true&characterEncoding=utf-8&useSSL=false"     jdbc_user => "root"     jdbc_password => "root"

    # jdbc連線mysql驅動的檔案目錄，可去官網下載:https://dev.mysql.com/downloads/connector/j/     jdbc_driver_library => "./config/mysql-connector-java-5.1.6.jar"     # the name of the driver class for mysql     jdbc_driver_class => "com.mysql.jdbc.Driver"     jdbc_paging_enabled => true     jdbc_page_size => "50000"

    jdbc_default_timezone =>"Asia/Shanghai"

    # mysql檔案, 也可以直接寫SQL語句在此處，如下：     statement => "select * from spider_car_resource where  to_days(created) = to_days(now())-1"     #statement_filepath => "./config/jdbc.sql"

    # 這裡類似crontab,可以定製定時操作，比如每分鐘執行一次同步(分 時 天 月 年)     schedule => "* * * * *"     #type => "jdbc"

    # 是否記錄上次執行結果, 如果為真,將會把上次執行到的 tracking_column 欄位的值記錄下來,儲存到 last_run_metadata_path 指定的檔案中     #record_last_run => true

    # 是否需要記錄某個column 的值,如果record_last_run為真,可以自定義我們需要 track 的 column 名稱，此時該引數就要為 true. 否則預設 track 的是 timestamp 的值.     use_column_value => true

    # 如果 use_column_value 為真,需配置此引數. track 的資料庫 column 名,該 column 必須是遞增的. 一般是mysql主鍵     tracking_column => "update_time"          tracking_column_type => "timestamp"

    last_run_metadata_path => "./logstash_capital_bill_last_id"

    # 是否清除 last_run_metadata_path 的記錄,如果為真那麼每次都相當於從頭開始查詢所有的資料庫記錄     clean_run => false

    #是否將 欄位(column) 名稱轉小寫     lowercase_column_names => false   } }

output {   elasticsearch {     hosts => "localhost:9200"     index => "test_spider_car_resource"     document_id => "%{id}"     template_overwrite => true   }

  # 這裡輸出除錯，正式執行時可以註釋掉   stdout {       codec => json_lines   }  }

執行命令

bin/logstash -f config/logstash-mysql-es.conf > logs/logstash.out &

注意：

1、sql不要有分號