应对sharding-jdbc结合mybatis实现分库分表功能分表的联合查询采用将mysql的数据同步到elasticsearch进行筛选

朱雀 2022-05-19 00:53 224阅读 0赞

### **应对sharding-jdbc结合mybatis实现分库分表功能  分表的联合查询采用将mysql的数据同步到elasticsearch进行筛选** ###

> ** （一）windows平台上[ElasticSearch5+logstash的logstash-input-jdbc实现mysql数据同步][ElasticSearch5_logstash_logstash-input-jdbc_mysql]**

**安装操作指南：[（1）][1]、[（2）][2]**

**其中windows目录展示如下：**

**版本控制：1. 需要jdk:1.8（1.8.0\_60）**

**                  2.ElasticSearch(5.6.3)**

**                  3.建议logstash版本与ElasticSearch版本号对应一致（5.6.3）**

**                  4.ruby-2.4.4-2-x64**

**[工具下载地址：][Link 1]**

**说明：**

**ElasticSearch配置head插件：**

1.head在es2.X及以前是作为es的一个插件存在的,可以直接通过plugin命令安装,5.x及以后需要作为一个单独的服务运行,基于nodejs环境及grunt启动服务

2.logstash + Logstash-input-jdbc来同步数据,logstash-input-jdbc插件是logstash 的一个个插件,使用ruby语言开发。所以要先安装ruby,也是为了好使用ruby中的gem安装插件，下载地址: [https://rubyinstaller.org/downloads/][https_rubyinstaller.org_downloads]

![70][]

> ## **核心：同步方式** ##

1、数据同步方式

全量同步与增量同步

全量同步是指全部将数据同步到es，通常是刚建立es，第一次同步时使用。增量同步是指将后续的更新、插入记录同步到es。

2. logstash-input-jdbc使用

官方文档地址

[https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html][https_www.elastic.co_guide_en_logstash_current_plugins-inputs-jdbc.html]

logstash目录下配置文件的结构：

![70 1][]

全量同步配置文件：jdbc.conf   + jdbc.sql + mysql-connector-java-5.1.9

增量同步配置文件：jdbc\_add.conf   + jdbc\_add.sql + mysql-connector-java-5.1.9 + station\_parameter.txt

jdbc..conf：

jdbc.sql：

select * from t_student_01

增量同步jdbc\_add.sql：

input {
        stdin {
        }
        jdbc {
          # mysql 数据库链接,test为数据库名
          jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test"
          # 用户名和密码
          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
        }
    }
    参数介绍:
    
    //是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
    record_last_run => true
    
    //是否需要记录某个column 的值,如果 record_last_run 为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
    use_column_value => true
    
    //如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的.比如：ID.
    tracking_column => MY_ID
    
    //指定文件,来记录上次执行到的 tracking_column 字段的值
    //比如上次数据库有 10000 条记录,查询完后该文件中就会有数字 10000 这样的记录,下次执行 SQL 查询可以从 10001 条处开始.
    //我们只需要在 SQL 语句中 WHERE MY_ID > :sql_last_value 即可. 其中 :sql_last_value 取得就是该文件中的值(10000).(！！！注意这里唯一不变就是 sql_last_value 是数据库表的主键的当前最大值)
    last_run_metadata_path => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\station_parameter.txt"
    
    
    //是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
    clean_run => false
    
    //是否将 column 名称转小写
    lowercase_column_names => false
    
    //存放需要执行的 SQL 语句的文件位置
    statement_filepath => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\jdbc.sql"

增量同步：jdbc\_add.sql:

select * from t_student_01 where id > :sql_last_value

增量同步：station\_parameter.txt（！注意  \#上一个sql\_last\_value值的存放文件路径, 起初必须要在文件中指定字段的初始值，其中“---”这符号是自己生成的）

![70 2][]

**最终效果： **

这个文件里记录上次执行到的 tracking\_column 字段的值,比如上次数据库有 10000 条记录,查询完后该文件中就会有数字 10000 这样的记录,下次执行 SQL 查询可以从 10001 条处开始,我们只需要在 SQL 语句中 WHERE MY\_ID > :sql\_last\_value 即可. 其中 :sql\_last\_value 取得就是该文件中的值。

> ## **遇到的问题：** ##

（1）在安装好ElasticSearch的head插件后总是用grunt server命令启动时总是报错：

![70 3][]

**解决方案： 命令：**

npm install grunt --save-dev

（2）增量同步中上述jdbc\_add.sql文件中的 记录tracking\_column 字段的值,  必须是

where id > :sql_last_value

（3）上述所有配置文件需要编码是ANSI编码方式（一般默认是UTF-8编码），否则出错

[ElasticSearch5_logstash_logstash-input-jdbc_mysql]: https://www.cnblogs.com/phpshen/p/6098333.html
[1]: https://blog.csdn.net/blq4411568/article/details/78453735
[2]: https://www.cnblogs.com/linsu/p/7837422.html
[Link 1]: https://download.csdn.net/download/caox_nazi/10548944
[https_rubyinstaller.org_downloads]: https://rubyinstaller.org/downloads/
[70]: /images/20220519/2f120417cc8c4bcdbb9e17b63ead668b.png
[https_www.elastic.co_guide_en_logstash_current_plugins-inputs-jdbc.html]: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html
[70 1]: /images/20220519/8d047ea415b543d18a180839646d041d.png
[70 2]: /images/20220519/2dbcedc387644652862735c07626ffc1.png
[70 3]: /images/20220519/246395804f7648849215938d7ee05e9d.png