logstash解決增量資料

input {
    stdin {
    }
    jdbc {
      # mysql 資料庫連結,test為資料庫名
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/datacube"
      # 使用者名和密碼
      jdbc_user => "root"
      jdbc_password => "root"
      # 驅動
      jdbc_driver_library => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\mysql-connector-java-5.1.9.jar"
      # 驅動類名
      jdbc_driver_class => "com.mysql.jdbc.Driver"

      #進行中文亂碼問題
      codec => plain { charset => "UTF-8"}
      #使用其它字段追蹤，而不是用時間
      use_column_value => true
      #追蹤的字段
      tracking_column => id
      record_last_run => true
     #上一個sql_last_value值的存放檔案路徑, 必須要在檔案中指定字段的初始值
     last_run_metadata_path => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\station_parameter.txt"
     #開啟分頁查詢
     jdbc_paging_enabled => true
     jdbc_page_size => 300
	  
      # 執行的sql 檔案路徑+名稱
      statement_filepath => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\jdbc.sql"
      # 設定監聽間隔  各字段含義（由左至右）分、時、天、月、年，全部為*預設含義為每分鐘都更新
	  schedule => "* * * * *"
      # 索引類型
	  type => "jdbc"

    }
}

filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}

output {
    elasticsearch {
	    # ES的IP位址及端口
        hosts => ["localhost:9200"]
		# 索引名稱
        index => "article"
		# 自增ID
        document_id => "%{id}"
    }
    stdout {
	    # JSON格式輸出
        codec => json_lines
    }
}