Logstash同步MySQL到ES

一、同步要点：
1、logstash每分钟执行一次
2、sql 语句必须带limit N，这里的N，在串行单并发执行时，建议为100000，分页每次50000
3、从MySQL从库同步数据

二、同步测试

1676340249422.jpg

三、负载情况
1、同步机器负载

image.png
image.png

2、ES磁盘读写

image.png

3、db机器负载

image.png

4、db延迟负载

image.png

四、Logstash配置文件

input {
    jdbc {
        # 设置 MySql/MariaDB 数据库url以及数据库名称
        jdbc_connection_string => "jdbc:mysql://10.28.54.130:3306/mydb?useUnicode=true&allowMultiQuerie=true&characterEncoding=utf-8&serverTimezone=UTC"
        # 用户名和密码
        jdbc_user => "root"
        jdbc_password => "xxxxx"
        # 数据库驱动mysql-connector-java-8.0.19.jar所在位置，可以是绝对路径或者相对路径
        jdbc_driver_library => "/data/dba_logstash/logstash-7.17.7/mysql/mysql-connector-java-5.1.49.jar"
        # 驱动类名
        #jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
         jdbc_driver_class => "org.gjt.mm.mysql.Driver"
         # 是否开启分页,ture为开启，我这里sql比较复杂所以就放弃使用这个，后面细讲
        jdbc_paging_enabled => true
         # 分页每页数量
        jdbc_page_size => "50000"
        # 设置时区
        jdbc_default_timezone =>"Asia/Shanghai"
        # 执行的sql文件路径
        #statement_filepath => "/data/dba_logstash/logstash-7.17.7/mysql/my.sql"
        #使用这个可以直接写sql语句，但是复杂的语句最好是写在文件内
        statement => "SELECT uhid,nickname as nick_name,mobile_ed,min_head_img_url,head_img_url,status,create_time,last_update_time,given_nickname,given_head_img_url,brief_intro,generations from new_user_info WHERE last_update_time > :sql_last_value ORDER
BY last_update_time limit 100000"
        # 设置定时任务间隔  含义：分、时、天、月、年，全部为*默认含义为每分钟跑一次任务
        schedule => "* * * * *"
        #是否需要记录某个字段值,如果为true,我们可以自定义要记录的数据库某个字段值，例如id或date字段。如果为false，记录的是上次执行的标记，默认是一个timestamp
        use_column_value => true
        #记录上次执行字段值路径。我们可以在sql语句中这么写：WHERE ID > :last_sql_value。其中 :sql_last_value 取得就是该文件中的值，这个last_id会以文件形式存在，上面截图有
        last_run_metadata_path => "/data/dba_logstash/logstash-7.17.7/mysql/last_id"
        #如果use_column_value为真,需配置此参数. 指定增量更新的字段名。当然该字段必须是递增的，比如id或date字段。
        tracking_column => "last_update_time"
        # tracking_column 对应字段的类型，只能选择timestamp或者numeric(数字类型)，默认numeric，所以可以不写这个配置
        tracking_column_type => "timestamp"
       #如果为true，每次会记录所更新的字段的值,并保存到 last_run_metadata_path 指定的文件中
        record_last_run => true
        # 是否清除 last_run_metadata_path 的记录，true则每次都从头开始查询所有的数据库记录
        clean_run => false
         # 是否将字段名称转小写。默认是true。这里注意Elasticsearch是区分大小写的
        lowercase_column_names => false
    }