想要实时搜索，最重要的是同步数据库

lsekfe

目前大部分mysql和elasticsearch同步机制使用的插件实现的，常用的插件包为：logstash-input-jdbc,go-mysql-elasticsearch, elasticsearch-jdbc、canal
　　插件优缺点对比
　　1. logstash-input-jdbc
　　logstash官方插件,集成在logstash中,下载logstash即可,通过配置文件实现mysql与elasticsearch数据同步
　　优点
　　·能实现mysql数据全量和增量的数据同步,且能实现定时同步。
　　·版本更新迭代快,相对稳定。
　　·作为ES固有插件logstash一部分,易用。
　　缺点
　　·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在。
　　·同步最短时间差为一分钟,一分钟数据同步一次,无法做到实时同步。
　　2、go-mysql-elasticsearch
　　go-mysql-elasticsearch 是国内作者开发的一款插件
　　优点
　　·能实现mysql数据增加,删除,修改操作的实时数据同步
　　缺点
　　·无法实现数据全量同步Elasticsearch
　　·仍处理开发、相对不稳定阶段
　　3、elasticsearch-jdbc
　　目前最新的版本是2.3.4，支持的ElasticSearch的版本为2.3.4, 未实践
　　优点
　　·能实现mysql数据全量和增量的数据同步.
　　缺点
　　·目前最新的版本是2.3.4，支持的ElasticSearch的版本为2.3.4
　　·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在.
　　mysql安装
　　安装依赖
　　yum search libaio  # 检索相关信息
　　yum install libaio # 安装依赖包
　　mysql是否安装
　　yum list installed | grep mysql
　　mysql卸载
　　yum -y remove mysql-libs.x86_64
　　mysql yum下载
　　wget -i -c http://dev.mysql.com/get/mysql57 ... e-el7-10.noarch.rpm
　　安装 MySQL
　　yum -y install mysql57-community-release-el7-10.noarch.rpm
　　yum -y install mysql-community-server
　　查看mysql安装位置
　　whereis mysql
　　启动mysql
　　systemctl start  mysqld.service
　　systemctl status mysqld.service
　　关闭mysql
　　systemctl stop  mysqld
　　查看密码
　　grep 'temporary password' /var/log/mysqld.log
　　mysql修改密码远程连接
　　SET PASSWORD = PASSWORD('/20as3SElksds0ew98');
　　GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '/20as3SElksds0ew98' WITH GRANT OPTION;
　　logstash-input-jdbc实现mysql数据库与elasticsearch同步
　　logstash5.x之后，集成了logstash-input-jdbc插件。安装logstash后通过命令安装logstash-input-jdbc插件

1. ./logstash-plugin install logstash-input-jdbc

复制代码

配置
　　在logstash/config文件夹下新建jdbc.conf,配置内容如下：

1. # 输入部分
   
2. 
   
3. 　　input {
   
4. 
   
5. 　　  stdin {}
   
6. 
   
7. 　　  jdbc {
   
8. 
   
9. 　　    # mysql数据库驱动
   
10. 
    
11. 　　    jdbc_driver_library => "../config/mysql-connector-java-5.1.30.jar"
    
12. 
    
13. 　　    jdbc_driver_class => "com.mysql.jdbc.Driver"
    
14. 
    
15. 　　    # mysql数据库链接，数据库名
    
16. 
    
17. 　　    jdbc_connection_string => "jdbc:mysql://localhost:3306/cmr"
    
18. 
    
19. 　　    # mysql数据库用户名，密码
    
20. 
    
21. 　　    jdbc_user => "root"
    
22. 
    
23. 　　    jdbc_password => "12345678"
    
24. 
    
25. 　　    # 设置监听间隔  各字段含义（分、时、天、月、年），全部为*默认含义为每分钟更新一次
    
26. 
    
27. 　　    schedule => "* * * * *"
    
28. 
    
29. 　　    # 分页
    
30. 
    
31. 　　    jdbc_paging_enabled => "true"
    
32. 
    
33. 　　    # 分页大小
    
34. 
    
35. 　　    jdbc_page_size => "50000"
    
36. 
    
37. 　　    # sql语句执行文件，也可直接使用 statement => 'select * from t_school_archives_fold create_time >=
    
38. 
    
39. 　　                                :sql_last_value order by create_time limit 200000'
    
40. 
    
41. 　　    statement_filepath => "/config/jdbc.sql"
    
42. 
    
43. 　　    # elasticsearch索引类型名
    
44. 
    
45. 　　    type => "t_employee"
    
46. 
    
47. 　　  }
    
48. 
    
49. 　　}
    
50. 
    
51. 　　# 过滤部分(不是必须项）
    
52. 
    
53. 　　filter {
    
54. 
    
55. 　　    json {
    
56. 
    
57. 　　        source => "message"
    
58. 
    
59. 　　        remove_field => ["message"]
    
60. 
    
61. 　　    }
    
62. 
    
63. 　　}
    
64. 
    
65. 　　# 输出部分
    
66. 
    
67. 　　output {
    
68. 
    
69. 　　    elasticsearch {
    
70. 
    
71. 　　        # elasticsearch索引名
    
72. 
    
73. 　　        index => "octopus"
    
74. 
    
75. 　　        # 使用input中的type作为elasticsearch索引下的类型名
    
76. 
    
77. 　　        document_type => "%{type}"   # <- use the type from each input
    
78. 
    
79. 　　        # elasticsearch的ip和端口号
    
80. 
    
81. 　　        hosts => "localhost:9200"
    
82. 
    
83. 　　        # 同步mysql中数据id作为elasticsearch中文档id
    
84. 
    
85. 　　        document_id => "%{id}"
    
86. 
    
87. 　　    }
    
88. 
    
89. 　　    stdout {
    
90. 
    
91. 　　        codec => json_lines
    
92. 
    
93. 　　    }
    
94. 
    
95. 　　}
    
96. 
    
97. 　　# 注: 使用时请去掉此文件中的注释，不然会报错

复制代码

在config 目录下新建jdbc.sql文件

1. select * from t_employee

复制代码

运行

1. cd logstash-6.4.2
   
2. 
   
3. 　　# 检查配置文件语法是否正确
   
4. 
   
5. 　　bin/logstash -f config/jdbc.conf --config.test_and_exit
   
6. 
   
7. 　　# 启动
   
8. 
   
9. 　　bin/logstash -f config/jdbc.conf --config.reload.automatic
   
10. 
    
11. 　　--config.reload.automatic: 会自动重新加载配置文件内容
    
12. 
    
13. 　　在kibana中创建索引后查看同步数据
    
14. 
    
15. 　　PUT octopus
    
16. 
    
17. 　　GET octopus/_search

复制代码

Canal实现mysql数据库与elasticsearch同步
　　mysql
　　修改/etc/my.cnf

1. 　log-bin=mysql-bin
   
2. 
   
3. 　　binlog-format=ROW  
   
4. 
   
5. 　　server-id=1

复制代码

创建授权

1. create user canal identified by 'Canal@2020!';   #创建canal账户
   
2. 
   
3. 　　grant select,replication slave,replication client on *.* to 'canal'@'%'; #授权canal账户查询和复制权限
   
4. 
   
5. 　　flush privileges;                                                        #刷新授权

复制代码

查看binlog是否正确启动

1. show variables like 'binlog_format%';

复制代码

创建需要同步的数据库

1. 　create database canal_testdb character set utf8;  
   
2. 
   
3. 　　CREATE TABLE canal_table (   
   
4. 
   
5. 　　#创建canal_table表，字段为 id age name address
   
6. 
   
7. 　　id int(11) NOT NULL,
   
8. 
   
9. 　　age int(11) NOT NULL,
   
10. 
    
11. 　　name varchar(200) NOT NULL,
    
12. 
    
13. 　　address varchar(1000) DEFAULT NULL,
    
14. 
    
15. 　　PRIMARY KEY (id)
    
16. 
    
17. 　　) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    
18. 
    
19. 　　INSERT INTO canal_testdb.canal_table(id, age, name, address) VALUES (1, 88, '小明', '测试');

复制代码

Elasticsearch参考第一节
　　部署Canal-deployer服务端
　　下载并解压

1. # 没有的话新建
   
2. 
   
3. 　　cd /usr/local/canal/canal-deployer/  
   
4. 
   
5. 　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.deployer-1.1.5-SNAPSHOT.tar.gz
   
6. 
   
7. 　　tar xf canal.deployer-1.1.5-SNAPSHOT.tar.gz

复制代码

修改配置文件 instance.properties

1. 　vim /usr/local/canal-deployer/conf/example/instance.properties
   
2. 
   
3. 　　#################################################
   
4. 
   
5. 　　## mysql serverId , v1.0.26+ will autoGen
   
6. 
   
7. 　　canal.instance.mysql.slaveId=3
   
8. 
   
9. 　　# enable gtid use true/false
   
10. 
    
11. 　　canal.instance.gtidon=false
    
12. 
    
13. 　　# position info
    
14. 
    
15. 　　canal.instance.master.address=127.0.0.1:3306
    
16. 
    
17. 　　canal.instance.master.journal.name=
    
18. 
    
19. 　　canal.instance.master.position=
    
20. 
    
21. 　　canal.instance.master.timestamp=
    
22. 
    
23. 　　canal.instance.master.gtid=
    
24. 
    
25. 　　# rds oss binlog
    
26. 
    
27. 　　canal.instance.rds.accesskey=
    
28. 
    
29. 　　canal.instance.rds.secretkey=
    
30. 
    
31. 　　canal.instance.rds.instanceId=
    
32. 
    
33. 　　# table meta tsdb info
    
34. 
    
35. 　　canal.instance.tsdb.enable=true
    
36. 
    
37. 　　#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_testdb
    
38. 
    
39. 　　#canal.instance.tsdb.dbUsername=canal
    
40. 
    
41. 　　#canal.instance.tsdb.dbPassword=canal
    
42. 
    
43. 　　#canal.instance.standby.address =
    
44. 
    
45. 　　#canal.instance.standby.journal.name =
    
46. 
    
47. 　　#canal.instance.standby.position =
    
48. 
    
49. 　　#canal.instance.standby.timestamp =
    
50. 
    
51. 　　#canal.instance.standby.gtid=
    
52. 
    
53. 　　# username/password
    
54. 
    
55. 　　canal.instance.dbUsername=canal
    
56. 
    
57. 　　canal.instance.dbPassword=Canal@2020!
    
58. 
    
59. 　　canal.instance.connectionCharset = UTF-8
    
60. 
    
61. 　　# enable druid Decrypt database password
    
62. 
    
63. 　　canal.instance.enableDruid=false
    
64. 
    
65. 　　#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
    
66. 
    
67. 　　# table regex
    
68. 
    
69. 　　canal.instance.filter.regex=.*\\..*
    
70. 
    
71. 　　# table black regex
    
72. 
    
73. 　　canal.instance.filter.black.regex=
    
74. 
    
75. 　　# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
    
76. 
    
77. 　　#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
    
78. 
    
79. 　　# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
    
80. 
    
81. 　　#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
    
82. 
    
83. 　　# mq config
    
84. 
    
85. 　　canal.mq.topic=example
    
86. 
    
87. 　　# dynamic topic route by schema or table regex
    
88. 
    
89. 　　#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
    
90. 
    
91. 　　canal.mq.partition=0
    
92. 
    
93. 　　# hash partition config
    
94. 
    
95. 　　#canal.mq.partitionsNum=3
    
96. 
    
97. 　　#canal.mq.partitionHash=test.table:id^name,.*\\..*
    
98. 
    
99. 　　#################################################

复制代码

　启动canal-deployer
　　因为canal-depaloyer由java开发，所以需要jdk环境，jdk版本需要大于1.5
　　yum install java-1.8.0-openjdk.x86_64 java-1.8.0-openjdk-devel.x86_64 -y
　　/usr/local/canal/canal-deployer/bin/startup.sh
　　查看日志及端口
　　tail -f /usr/local/canal/logs/example/example.log
　　canal-deployer默认监听三个端口,11110、11111、11112
　　11110：为admin管理端口
　　11111：为canal deployer 服务器占用的端口
　　11112：为指标下拉端口
　　部署Canal-adapter客户端
　　下载并解压

1. 　cd /usr/local/canal/canal-adapter/  
   
2. 
   
3. 　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.adapter-1.1.5-SNAPSHOT.tar.gz
   
4. 
   
5. 　　tar xf canal.adapter-1.1.5-SNAPSHOT.tar.gz

复制代码

添加mysql8.0.18连接器

1. 　cd /usr/local/canal/canal-adapter/lib/
   
2. 
   
3. 　　wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.18/mysql-connector-java-8.0.18.jar
   
4. 
   
5. 　　chmod 777 /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar                  #权限修改与其它lib库一致
   
6. 
   
7. 　　chmod +st /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar

复制代码

　修改application.yml

1. server:
   
2. 
   
3. 　　  port: 8081
   
4. 
   
5. 　　spring:
   
6. 
   
7. 　　  jackson:
   
8. 
   
9. 　　    date-format: yyyy-MM-dd HH:mm:ss
   
10. 
    
11. 　　    time-zone: GMT+8
    
12. 
    
13. 　　    default-property-inclusion: non_null
    
14. 
    
15. 　　canal.conf:
    
16. 
    
17. 　　  mode: tcp # kafka rocketMQ
    
18. 
    
19. 　　  canalServerHost: 127.0.0.1:11111
    
20. 
    
21. 　　#  zookeeperHosts: slave1:2181
    
22. 
    
23. 　　#  mqServers: 127.0.0.1:9092 #or rocketmq
    
24. 
    
25. 　　#  flatMessage: true
    
26. 
    
27. 　　  batchSize: 500
    
28. 
    
29. 　　  syncBatchSize: 1000
    
30. 
    
31. 　　  retries: 0
    
32. 
    
33. 　　  timeout:
    
34. 
    
35. 　　  srcDataSources:
    
36. 
    
37. 　　    defaultDS:
    
38. 
    
39. 　　      url: jdbc:mysql://127.0.0.1:3306/canal_testdb?useUnicode=true
    
40. 
    
41. 　　      username: canal
    
42. 
    
43. 　　      password: Canal@2020!
    
44. 
    
45. 　　  canalAdapters:
    
46. 
    
47. 　　  - instance: example # canal instance Name or mq topic name
    
48. 
    
49. 　　    groups:
    
50. 
    
51. 　　    - groupId: g1
    
52. 
    
53. 　　      outerAdapters:
    
54. 
    
55. 　　      - name: logger
    
56. 
    
57. 　　      - name: es7
    
58. 
    
59. 　　        hosts: 192.168.0.200:9300,192.168.0.200:8200
    
60. 
    
61. 　　        properties:
    
62. 
    
63. 　　          mode: rest # or rest
    
64. 
    
65. 　　          # security.auth: test:123456 #  only used for rest mode
    
66. 
    
67. 　　          cluster.name: my-es

复制代码

修改适配器映射文件

1. vim /usr/local/canal/canal-adapter/conf/es7/mytest_user.yml
   
2. 
   
3. 　　dataSourceKey: defaultDS                                        #指定在application.yml文件中srcDataSources源数据源自定义的名称
   
4. 
   
5. 　　destination: example                                            #cannal的instance或者MQ的topic，我们是把数据同步至es，所以不用修改，也用不到此处
   
6. 
   
7. 　　groupId: g1                                                     #对应MQ模式下的groupId, 只会同步对应groupId的数据
   
8. 
   
9. 　　esMapping:                                                      #es中的Mapping设置
   
10. 
    
11. 　　  _index: canal_tsdb                                            #指定索引名称
    
12. 
    
13. 　　  _id: _id                                                      #指定文档id，_id 此值则由es自动分配文档ID
    
14. 
    
15. 　　  sql: "select a.id as _id,a.age,a.name,a.address from canal_table a"        #sql映射
    
16. 
    
17. 　　  etlCondition: "where a.c_time>={}"                            #etl的条件参数
    
18. 
    
19. 　　  commitBatch: 3000                                             #提交批大小

复制代码

　Elasticsearch创建索引

1. POST canal_tsdb/_doc
   
2. 
   
3. 　　{
   
4. 
   
5. 　　    "mappings":{
   
6. 
   
7. 　　        "_doc":{
   
8. 
   
9. 　　            "properties":{
   
10. 
    
11. 　　                "age":{
    
12. 
    
13. 　　                    "type":"long"
    
14. 
    
15. 　　                },
    
16. 
    
17. 　　                "name":{
    
18. 
    
19. 　　                    "type":"text"
    
20. 
    
21. 　　                },
    
22. 
    
23. 　　                "address":{
    
24. 
    
25. 　　                    "type":"text"
    
26. 
    
27. 　　                }
    
28. 
    
29. 　　            }
    
30. 
    
31. 　　        }
    
32. 
    
33. 　　    }
    
34. 
    
35. 　　

复制代码

　启动Canal-adapter并写入数据
　　/usr/local/canal/canal-adapter/bin/startup.sh
　　tail -f /usr/local/canal/canal-adapter/logs/adapter/adapter.log
　　在MySQL再次插入一条数据并查看日志
　　INSERT INTO canal_tsdb.canal_table(id, age, name, address) VALUES (2, 88, '小明', '测试');
　　查看Canal-deployer服务端日志
　　tail -f /usr/local/canal/canal-deployer/logs/example/meta.log
　　在es里面可以看到数据
　　部署Canal Admin
　　canal-admin的限定依赖：
　　1.MySQL，用于存储配置和节点等相关数据
　　2.canal版本，要求>=1.1.4 (需要依赖canal-server提供面向admin的动态运维管理接口)
　　Canal Admin下载并解压

1. 　mkdir /usr/local/canal/canal-admin
   
2. 
   
3. 　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.admin-1.1.5-SNAPSHOT.tar.gz
   
4. 
   
5. 　　tar xf canal.admin-1.1.5-SNAPSHOT.tar.gz

复制代码

application.yml

1. server:
   
2. 
   
3. 　　  port: 8089                                    #Canal Admin监听端口
   
4. 
   
5. 　　spring:
   
6. 
   
7. 　　  jackson:
   
8. 
   
9. 　　    date-format: yyyy-MM-dd HH:mm:ss            #时间格式
   
10. 
    
11. 　　    time-zone: GMT+8                            #时区
    
12. 
    
13. 　　spring.datasource:                              #数据库信息
    
14. 
    
15. 　　  address: 192.168.0.200:8809                  #指定Canal Admin所使用的数据库地址及端口
    
16. 
    
17. 　　  database: canal_manager                       #指定数据库名称
    
18. 
    
19. 　　  username: cadmin                              #指定数据库账户
    
20. 
    
21. 　　  password: Cadmin@2020!                        #指定数据库密码
    
22. 
    
23. 　　  driver-class-name: com.mysql.jdbc.Driver      #指定数据库驱动
    
24. 
    
25. 　　  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
    
26. 
    
27. 　　  hikari:
    
28. 
    
29. 　　    maximum-pool-size: 30
    
30. 
    
31. 　　    minimum-idle: 1
    
32. 
    
33. 　　canal:                                          #Canal UI界面默认账号密码
    
34. 
    
35. 　　  adminUser: admin
    
36. 
    
37. 　　  adminPasswd: admin

复制代码

创建数据库及授权用户

1. create database canal_manager character set utf8;
   
2. 
   
3. 　　create database canal_manager character set utf8;
   
4. 
   
5. 　　create user cadmin identified by 'Cadmin@2020!';
   
6. 
   
7. 　　grant all on canal_manager.* to 'cadmin'@'%';
   
8. 
   
9. 　　flush privileges;

复制代码

/usr/local/canal/canal-admin/conf/canal_manager.sql;数据库数据导入
　　启动
　　/usr/local/canal/canal-admin/bin/startup.sh