将mysql数据同步到ES6.4（全量+增量）

input {
    stdin {
    }
    jdbc {
      # 连接的数据库地址和哪一个数据库，指定编码格式，禁用SSL协议，设定自动重连
      jdbc_connection_string => "jdbc:mysql://数据库地址:端口号/数据库名?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
      # 你的账户密码
      jdbc_user => "账号"
      jdbc_password => "密码"
      # 连接数据库的驱动包，建议使用绝对地址
      jdbc_driver_library => "mysql/mysql-connector-java-5.1.45-bin.jar"
      # 这是不用动就好
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"

   #处理中文乱码问题
      codec => plain { charset => "UTF-8"}
   #使用其它字段追踪，而不是用时间
    use_column_value => true
   #追踪的字段      
    tracking_column => testid    
　　#如果你追踪的字段是时间类型
　　#tracking_column_type => "timestamp"  
    record_last_run => true     
   #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值     
    last_run_metadata_path => "mysql/station_parameter.txt"

    jdbc_default_timezone => "Asia/Shanghai"
 
    statement_filepath => "mysql/jdbc.sql"
      

    #是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
    clean_run => false

      # 这是控制定时的，重复执行导入任务的时间间隔，第一位是分钟
      schedule => "* * * * *"
      type => "jdbc"
    }
}


filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}
 
 
output {
    elasticsearch {
        # 要导入到的Elasticsearch所在的主机
        hosts => "192.168.105.180:9200"
        # 要导入到的Elasticsearch的索引的名称
        index => "db_anytest"
        # 类型名称（类似数据库表名）
        document_type => "table_anytest"
        # 主键名称（类似数据库主键）
        document_id => "%{testid}"
        
    }

    stdout {
        # JSON格式输出
        codec => json_lines
    }
    
}