梦神十夜的小站

Logstash是ES全家桶中同步数据的工具，可以直连数据库进行数据同步
使用场景为定时同步数据库表数据，通过定时器设置同步频率，最低频率1分钟1次，无法做到绝对实时。需要绝对实时，可以使用file的输入源方式。
input {
    stdin {}
    jdbc {
        type => "jdbc"
         # 数据库连接地址
        jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/TestDB?characterEncoding=UTF-8&autoReconnect=true""
         # 数据库连接账号密码；
        jdbc_user => "username"
        jdbc_password => "pwd"
         # MySQL依赖包路径；
        jdbc_driver_library => "mysql/mysql-connector-java-5.1.34.jar"
         # the name of the driver class for mysql
        jdbc_driver_class => "com.mysql.jdbc.Driver"
         # 数据库重连尝试次数
        connection_retry_attempts => "3"
         # 判断数据库连接是否可用，默认false不开启
        jdbc_validate_connection => "true"
         # 数据库连接可用校验超时时间，默认3600S
        jdbc_validation_timeout => "3600"
         # 开启分页查询（默认false不开启）；
        jdbc_paging_enabled => "true"
         # 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
        jdbc_page_size => "500"
         # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
         # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
         # statement_filepath => "mysql/jdbc.sql"
        statement => "SELECT KeyId,TradeTime,OrderUserName,ModifyTime FROM `DetailTab` WHERE ModifyTime>= :sql_last_value order by ModifyTime asc"
         # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
        lowercase_column_names => false
         # Value can be any of: fatal,error,warn,info,debug，默认info；
        sql_log_level => warn
         #
         # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
        record_last_run => true
         # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
        use_column_value => true
         # 需要记录的字段，用于增量同步，需是数据库字段
        tracking_column => "ModifyTime"
         # Value can be any of: numeric,timestamp，Default value is "numeric"
        tracking_column_type => timestamp
         # record_last_run上次数据存放位置；
        last_run_metadata_path => "mysql/last_id.txt"
         # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
        clean_run => false
         #
         # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
        schedule => "* * * * *"
    }
}
filter {
  if [del_ind] == 1 {
    mutate{ add_field => {"[@metadata][action]" => "delete"}}
  }else {
    mutate{ add_field => {"[@metadata][action]" => "index"}}
  }
}
output {
    elasticsearch {
     # host => "192.168.1.1"
     # port => "9200"
     # 配置ES集群地址
    hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
     # 索引名字，必须小写
    index => "consumption-%{type}-%{+YYYY.MM.dd}"
     # 数据唯一索引（建议使用数据库KeyID）
    document_id => "%{KeyId}"
        action => "%{[@metadata][action]}"
    }
    stdout {
        codec => json_lines
    }
}
梦神十夜的小站

LOGSTASH单表同步
