logstash 整合 es

vim jdbc.conf

input {
      stdin {
       }
      jdbc {
        # 连接的数据库地址和哪一个数据库，指定编码格式，禁用SSL协议，设定自动重连
        jdbc_connection_string => "jdbc:mysql://192.168.1.105:3306/logstash_data?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
        jdbc_user => "root"
        jdbc_password => "root"
        # 下载连接数据库的驱动包，建议使用绝对地址
       jdbc_driver_library => "/opt/logstash-7.6.2/config/mysql-connector-java-8.0.11.jar"

       jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
       jdbc_paging_enabled => "true"
       jdbc_page_size => "50000"
       codec => plain { charset => "UTF-8"}

        #使用其它字段追踪，而不是用时间
      use_column_value => true   #这里如果是用时间追踪比如：数据的更新时间或创建时间等和时间有关的这里一定不能是true, 切记切记切记，我是用update_time来追踪的
        #追踪的字段
     tracking_column => update_date
     tracking_column_type => "timestamp"
     record_last_run => true
     #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值  这里说是必须指定初始值，我没指定默认是1970-01-01 08：00：00
     last_run_metadata_path => "./logstash_jdbc_last_run"  //这里的lastrun文件夹和.logstash_jdbc_last_run是自己创建的

      jdbc_default_timezone => "Asia/Shanghai"   //设置时区
      #statement => SELECT * FROM logstash  WHERE update_date >= :last_sql_value  //这里要说明一下如果直接写sql语句，前面这种写法肯定不对的
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　，加上引号也试过也不对，所以我直接写在jdbc.sql文件中
      #statement_filepath => "/usr/local/Cellar/logstash/6.5.4/bin/jdbc.sql"


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

       # 这是控制定时的，重复执行导入任务的时间间隔，第一位是分钟 不设置就是1分钟执行一次
       schedule => "* * * * *"
       type => "std"
     }
 }

 filter {

    json {

        source => "message"

        remove_field => ["message"]

    }

}

output {

    elasticsearch {

        # 要导入到的Elasticsearch所在的主机

        hosts => "127.0.0.1:9200"

        # 要导入到的Elasticsearch的索引的名称

        index => "product_index"

        # 类型名称（类似数据库表名）

        #document_type => ""

        # 主键名称（类似数据库主键）

        document_id => "%{id}"
        #是否使用模板，开启效率更高
        template_overwrite => true
    }

    stdout {

        # JSON格式输出

        codec => json_lines

    }
}
    

./bin/logstash ./config/jdbc.conf

logstash主目录下：vim jdbc.conf

input {
elasticsearch {
hosts => ["10.128.120.171", "10.128.120.172", "10.128.120.173", "10.128.120.179", "10.128.120.235"]
# user => "*******"
# password => "*********"
index => "*"
size => 1000
scroll => "1m"
}
}
filter {
mutate {
remove_field => ["@timestamp", "@version"]
}
}
output {
elasticsearch {
hosts => ["10.13.133.121", "10.13.133.122", "10.13.133.123", "10.13.133.124", "10.13.133.125"]
# user => "********"
# password => "**********"
index => "%{[@metadata][_index]}"
}
}

logstash -f jdbc.conf