搭建 canal 监控mysql数据到 elasticsearch 中

需求:

要将 MySQL 数据库 info 中的 notice 和 result 表的增、删、改操作同步到 Elasticsearch 的 notice 和 result 索引，您需要正确配置 MySQL、Canal 、Canal Adapter 、 Elasticsearch 和 kibana

一、搭建mysql

博客地址

结合上面的博客地址的安装方式，使用下面的配置文件即可

mysql配置文件，我使用的是下面这个

vim /etc/my.cnf

[mysql] default-character-set=utf8 socket=/var/lib/mysql/mysql.sock  [mysqld] log-bin=mysql-bin binlog-format=ROW expire-logs-days=7  # 可选，设置 Binlog 日志保留天数 # binlog-do-db=info  # 仅记录 info 数据库的 Binlog  bind-address = 0.0.0.0 #skip-name-resolve # （注意这里我端口修改为了3066） port = 3066  socket=/var/lib/mysql/mysql.sock basedir=/usr/local/mysql datadir=/data/mysql max_connections=5000 character-set-server=UTF8MB4 default-storage-engine=MyISAM lower_case_table_names=1 sql-mode=STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION sql_mode=STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION  server-id=1 innodb-file-per-table=1 wait_timeout=2880 interactive_timeout=2880 innodb_buffer_pool_size = 10G innodb_log_file_size= 1G key_buffer_size = 1536M max_allowed_packet = 1024M table_open_cache = 500 tmp_table_size = 2048M max_heap_table_size = 2048M sort_buffer_size = 16M read_buffer_size = 16M read_rnd_buffer_size = 32M myisam_sort_buffer_size = 128M thread_cache_size = 8  #relay_log=mall-mysql-relay-bin #relay_log_index=relay-log.index  [mysqld_safe] log-error=/var/log/mysql/mysql.log 

安装完成之后登录mysql 创建一个 canal 用户  CREATE USER 'canal'@'%' IDENTIFIED BY '123456'; GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;  允许root远程访问 CREATE USER 'root'@'%' IDENTIFIED WITH mysql_native_password BY '123456'; GRANT ALL ON *.* TO 'root'@'%'; FLUSH PRIVILEGES; 

二、安装Elasticsearch

操作系统优化

1 设置打开的文件句柄数和线程数 vim /etc/security/limits.conf  # 添加 # soft：软限制；hard：硬限制 # nproc：单个用户可打开的进程最大数 # nofile：单个进程打开文件最大数 # as：地址空间限制(unlimited：无限) # fsize：最大文件大小 # memlock：最大锁定内存地址空间 *               soft    nproc           65536 *               hard    nproc           65536 *               soft    nofile          65536 *               hard    nofile          65536 *               -       as              unlimited *               -       fsize           unlimited *               -       memlock         unlimited  2 关闭 swap 交换空间 swapoff -a && sed -i '/swap/s/^.*$/#&/' /etc/fstab  3 设置虚拟内存大小和 TCP 超时重传次数 vim /etc/sysctl.conf  # 添加 vm.max_map_count=262144 net.ipv4.tcp_retries2=5 net.core.somaxconn = 1024 vm.overcommit_memory = 1 # 默认情况下 TCP keepalive 时间为 60 秒，超时重传 15 次。  sysctl -p 

开始安装 Elasticsearch

下载&解压 Elasticsearch

wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.0-linux-x86_64.tar.gz tar -xf elasticsearch-8.15.0-linux-x86_64.tar.gz -C /usr/local/  useradd -u 9200 esuser mkdir -p /data/elasticsearch/{data,logs,temp} chown -R esuser:esuser  /data/elasticsearch/ /usr/local/elasticsearch-8.15.0/ cd /usr/local/elasticsearch-8.15.0/ 

修改配置文件

vim config/elasticsearch.yml

cluster.name: es-dev #集群名称 node.name: es #节点名称  #######----------这个配置只需要在kibana节点安装 ，且需要使用堆栈监测功能---------####### node.roles: [master,data,ingest, ml, transform, remote_cluster_client] #######----------如果使用堆栈功能，需要把安全认证关闭------------------#######  path.data: /data/elasticsearch/data # 数据存储位置 path.logs: /data/elasticsearch/logs #日志存储位置 network.host: 0.0.0.0 #允许连接IP # 允许跨域 http.port: 9200 # 网页访问端口 transport.profiles.default.port: 9300 http.cors.enabled: true  http.cors.allow-origin: "*" http.cors.allow-headers: "*" #http.cors.allow-methods: "GET" cluster.initial_master_nodes: ["es"] action.destructive_requires_name: false discovery.seed_hosts: ["192.168.174.136:9300"] # 集群成员  #如果关闭安全认证需要使用以下方法 xpack.security.enabled: false xpack.security.transport.ssl.enabled: false  #关闭geoip ingest.geoip.downloader.enabled: false xpack.monitoring.collection.enabled: true 

vim config/jvm.options

将注释删除，改为当前自己需要的内存大小，作者当前内存是16G，所以改为4G大小 1/4即可 -Xms4g -Xmx4g 

下载中文分词器

./bin/elasticsearch-plugin install https://get.infini.cloud/elasticsearch/analysis-ik/8.15.0 中途输入 y 下载完成会在 plugins 目录生成analysis-ik 目录 chown -R  esuser.esuser /usr/local/elasticsearch-8.15.0/ 

启动 elasticsearch

runuser -l esuser -c "/usr/local/elasticsearch-8.15.0/bin/elasticsearch -d"  # 查看日志 tail -f /data/elasticsearch/logs/es-dev.log 

三 、安装kibana

wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.0-linux-x86_64.tar.gz cd /usr/local/kibana-8.15.0/ 

修改配置
vim config/kibana.yml

server.port: 5601  server.host: "0.0.0.0" ##填本机IP或者 0.0.0.0 都可以，最好写本机IP  server.name: "devops-kibana" ## name 名称可以随便指定  ### es集群配置 elasticsearch.hosts: ["http://192.168.174.136:9200"]  pid.file: /usr/local/kibana-8.15.0/kibana.pid  elasticsearch.requestTimeout: 99999  i18n.locale: "zh-CN"   #---------------------#####------------------------- #如果高版本需要配置如下两个参数   用户名密码为 elasticsearch 安全证书用户密码  #如果没有生成证书认证，可不加如下两个参数 elasticsearch.username: "kibana" elasticsearch.password: "devops@123" 

启动kibana

useradd -u 5601 kibana chown -R kibana:kibana /usr/local/kibana-8.15.0/ nohup /usr/local/kibana-8.15.0/bin/kibana --allow-root > /var/log/kibana.log &  #查看日志 tail -f /var/log/kibana.log 

启动完成之后准备创建mysql测试数据

登录到mysql中， # 创建要监听的数据库 create database info;  # 创建一个名为 result 的表，并包含 id、user_id、score、created_at 和 updated_at 字段。 CREATE TABLE result ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, score DECIMAL(5, 2) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );  # 插入5 条数据 INSERT INTO result (user_id, score, created_at, updated_at) VALUES (1, 95.75, '2024-08-01 10:00:00', '2024-08-01 10:00:00'), (2, 88.50, '2024-08-01 10:05:00', '2024-08-01 10:05:00'), (3, 76.20, '2024-08-01 10:10:00', '2024-08-01 10:10:00'), (4, 82.00, '2024-08-01 10:15:00', '2024-08-01 10:15:00'), (5, 91.30, '2024-08-01 10:20:00', '2024-08-01 10:20:00'),  创建一个名为 notice 的表，并包含id, title, content, created_at, updated_at 字段。  CREATE TABLE notice ( id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255) NOT NULL, content TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );  id 列是自增的主键，title 是一个不允许为空的字符串，content 是可选的文本，created_at 和 updated_at 列处理时间戳  # 要插入 5 条数据到 notice 表中 ('Title 1', 'Content for notice 1', NOW(), NOW()), ('Title 2', 'Content for notice 2', NOW(), NOW()), ('Title 3', 'Content for notice 3', NOW(), NOW()), ('Title 4', 'Content for notice 4', NOW(), NOW()), ('Title 5', 'Content for notice 5', NOW(), NOW()), 

创建索引

登录到 kibana中 – 点击 开发工具，复制下面的命令，

创建 notice索引，复制到控制台，点击运行

PUT /notice {   "settings": {     "number_of_shards": 1,     "number_of_replicas": 1   },   "mappings": {     "properties": {       "id": {         "type": "keyword"  // 使用 keyword 类型，适合精确匹配       },       "title": {         "type": "text",    // 使用 text 类型，支持全文搜索         "analyzer": "standard" // 使用标准分析器       },       "content": {         "type": "text",    // 使用 text 类型，支持全文搜索         "analyzer": "standard" // 使用标准分析器       },       "created_at": {         "type": "date",    // 使用 date 类型，处理日期和时间         "format": "yyyy-MM-dd'T'HH:mm:ss" // 日期格式       },       "updated_at": {         "type": "date",    // 使用 date 类型，处理日期和时间         "format": "yyyy-MM-dd'T'HH:mm:ss" // 日期格式       }     }   } } 

创建 result 索引，复制到控制台，点击运行

PUT /result {   "settings": {     "number_of_shards": 1,     "number_of_replicas": 1   },   "mappings": {     "properties": {       "user_id": {         "type": "integer"       },       "score": {         "type": "float"       },       "created_at": {         "type": "date",         "format": "yyyy-MM-dd HH:mm:ss"       },       "updated_at": {         "type": "date",         "format": "yyyy-MM-dd HH:mm:ss"       }     }   } }  

注释：
Shards 和 Replicas：number_of_shards 定义了分片的数量，number_of_replicas 定义了每个主分片的副本数。根据你的数据量和需求进行调整。
Mappings：在创建索引时定义的数据类型对数据的索引和查询性能有很大影响，因此请确保使用合适的数据类型。

查看索引

创建索引后，你可以查看索引的详细信息以确认设置和映射是否正确：  curl -X GET "localhost:9200/result"	 curl -X GET "localhost:9200/notice"	 

向索引中添加文档

创建索引后，你可以向 notice 索引中添加文档。以下是一个示例： 

POST /notice/_doc/1 {   "id": "1",   "title": "Sample Notice Title",   "content": "This is the content of the notice.",   "created_at": "2024-08-29T08:00:00",   "updated_at": "2024-08-29T08:00:00" }  

查询索引中的数据

要查询 notice 索引中的数据，可以使用如下的查询请求： 

GET /notice/_search {   "query": {     "match_all": {}   } }  

四、下载安装 Canal

wget https://github.com/alibaba/canal/releases/download/canal-1.1.8-alpha-2/canal.adapter-1.1.8-SNAPSHOT.tar.gz wget https://github.com/alibaba/canal/releases/download/canal-1.1.8-alpha-2/canal.deployer-1.1.8-SNAPSHOT.tar.gz # 下面这两个安装包可有可无 wget https://github.com/alibaba/canal/releases/download/canal-1.1.8-alpha-2/canal.admin-1.1.8-SNAPSHOT.tar.gz wget https://github.com/alibaba/canal/releases/download/canal-1.1.8-alpha-2/canal.example-1.1.8-SNAPSHOT.tar.gz 

解压canal 与 canal-adapter

mkdir /usr/local/canal.adapter mkdir /usr/local/canal tar -xf canal.adapter-1.1.8-SNAPSHOT.tar.gz -C /usr/local/canal.adapter tar -xf canal.deployer-1.1.8-SNAPSHOT.tar.gz -C /usr/local/canal 

修改配置

cd /usr/local/canal/  vim conf/example/instance.properties 

################################################# ## mysql serverId , v1.0.26+ will autoGen # 这个值随意修改，但是不能和 mysql 主库的server-id值一样 canal.instance.mysql.slaveId=1001     # enable gtid use true/false canal.instance.gtidon=false  # position info # mysql 库的 IP地址和端口（注意这里我端口修改为了3066） canal.instance.master.address=127.0.0.1:3066 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= canal.instance.master.gtid=  # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId=  # table meta tsdb info canal.instance.tsdb.enable=true #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal  #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = #canal.instance.standby.gtid=  # username/password # mysql 账户与密码 canal.instance.dbUsername=root canal.instance.dbPassword=123456 canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==  # table regex #canal.instance.filter.regex=.*\\..* # 监听 info 库的notice表与result表 canal.instance.filter.regex=info\\.notice,info\\.result # table black regex # 不监听mysql库中以slave_开头的表 canal.instance.filter.black.regex=mysql\\.slave_.* # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch  # mq config # 这个默认即可 canal.mq.topic=example # dynamic topic route by schema or table regex #canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..* canal.mq.partition=0 # hash partition config #canal.mq.enableDynamicQueuePartition=false #canal.mq.partitionsNum=3 #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6 #canal.mq.partitionHash=test.table:id^name,.*\\..* # # multi stream for polardbx canal.instance.multi.stream.on=false ################################################# 

vim  conf/canal.properties 因为我们要将监控到的数据传到ES，所以修改地方比较少 

canal.serverMode = tcp canal.instance.tsdb.enable = true # 集群的配置只要把H2改为mysql，因为要进行元数据管理。 #canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:} #canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3066/canal_manager canal.instance.tsdb.dbUsername = root canal.instance.tsdb.dbPassword = 123456  #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml  

配置文件中注释的中文备注，记得观察。

启动 canal

./bin/startup.sh   # 查看日志 tail -f ./logs/canal/canal.log  输出如下及正常。 

tail -f logs/example/example.log （无报错及正常）  如果报错如下： 

[main] WARN o.s.context.support.ClassPathXmlApplicationContext - Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'tableMetaTSDB' defined in class path resource [spring/tsdb/h2-tsdb.xml]: Cannot resolve reference to bean 'metaHistoryDAO' while setting bean property 'metaHistoryDAO'; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'metaHistoryDAO' defined in class path resource [spring/tsdb/h2-tsdb.xml]: Cannot resolve reference to bean 'sqlSessionFactory' while setting bean property 'sqlSessionFactory'; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'sqlSessionFactory' defined in class path resource [spring/tsdb/h2-tsdb.xml]: Cannot resolve reference to bean 'dataSource' while setting bean property 'dataSource'; nested exception is org.springframework.beans.factory.CannotLoadBeanClassException: Cannot find class [com.alibaba.druid.pool.DruidDataSource] for bean with name 'dataSource' defined in class path resource [spring/tsdb/h2-tsdb.xml]; nested exception is java.lang.ClassNotFoundException: com.alibaba.druid.pool.DruidDataSource 

处理 将druid的jar包放在lib目录就可以了。druid-1.2.22.jar测试通过 下载地址 ：https://repo1.maven.org/maven2/com/alibaba/druid/1.2.22/ 

五、安装 Canal.adapter

cd /usr/local/canal.adapter vim application.yml 将多余的删除，取其精华即可 

server:   port: 8081 spring:   jackson:     date-format: yyyy-MM-dd HH:mm:ss     time-zone: GMT+8     default-property-inclusion: non_null canal.conf:   mode: tcp #tcp kafka rocketMQ rabbitMQ   flatMessage: true   zookeeperHosts:   syncBatchSize: 1000   retries: 0   timeout:   accessKey:   secretKey:   consumerProperties:     # canal tcp consumer     # canal.tcp.server.host需要修改     canal.tcp.server.host: 192.168.174.136:11111     canal.tcp.zookeeper.hosts:     canal.tcp.batch.size: 500     canal.tcp.username:     canal.tcp.password:   srcDataSources:     defaultDS:       # url,username,password需要修改（注意，这里端口我修改过） info是库名       url: jdbc:mysql://192.168.174.136:3066/info?useUnicode=true       username: root       password: 123456   canalAdapters:   - instance: example # canal instance Name or mq topic name     groups:     - groupId: gl        outerAdapters:       - name: logger       # name需要修改       # 这个表示我们使用的是哪个适配器，es8 表示使用的是 es8 适配器，其他的可以参考解压后的 conf 下面的目录名称       - name: es8         # hosts需要修改（注意，要加上http://）         hosts: http://127.0.0.1:9200 # 127.0.0.1:9200 for rest mode         properties:           mode: rest           # security.auth: test:123456 #  only used for rest mode           # cluster.name需要修改 ES集群名称           cluster.name: es-dev 

适配器配置

在上面的启动器的配置中我们已经配置了 ES8 作为适配器，那具体要同步的是哪张表， 以及对应的 ES中是索引是哪个怎么配置呢？这些配置就放在适配器的配置里面， 每一个适配器的配置都是一个想要同步到 ES 的模板配置。

因为我使用的es8适配器，所以进到es8中 cd conf/es8 cp -v mytest_user.yml esMappingNotice.yml rm -rf biz_order.yml  customer.yml   mytest_user.yml vim esMappingNotice.yml 

dataSourceKey: defaultDS destination: example groupId: gl esMapping:   _index: notice   _id: _id   _type: _doc   upsert: true   sql: " SELECT         c.id AS _id,         c.title AS title,         c.content AS content,         DATE_FORMAT (c.created_at, '%Y-%m-%dT%H:%i:%s') AS created,         DATE_FORMAT (c.updated_at, '%Y-%m-%dT%H:%i:%s') AS updated FROM         notice AS c "   commitBatch: 3000 

注释： dataSourceKey: defaultDS destination: example outerAdapterKey: es-key groupId: g1 上面的几个配置，都需要跟启动器里面的配置保持一致。  esMapping：该配置是表示的是如何将 MySQL 的数据同步到 ES 中，配置比较复杂，其中 _index 表示 ES 的索引（需要提前创建）； _id 和 pk 二选一配置，表示使用查询出来的哪个字段作为唯一值； upsert 表示对应主键的数据不存在的时候执行插入动作，存在的时候执行更新动作； sql：表示要同步的数据，这个的 SQL 形式要求会比较严格 而且 _id 必须要加别名，我索性把所有都改个别名  commitBatch: 3000 设置了每次批量提交的记录数量为 3000。这意味着每当 canal.adapter 收集到 3000 条记录时， 会将这些记录批量提交到 Elasticsearch。确保这个批量大小适合你的数据量和 Elasticsearch 的处理能力， 以避免超时或性能问题。如果你遇到性能瓶颈，可以尝试调整这个参数值，增大或减小批量大小来优化性能。 

vim esMappingResult.yml

dataSourceKey: defaultDS destination: example groupId: gl esMapping:   _index: result   _id: _id   _type: _doc   upsert: true   sql: " SELECT         c.id AS _id,         c.user_id AS userid,         c.score AS score,         DATE_FORMAT (c.created_at, '%Y-%m-%dT%H:%i:%s') AS created,         DATE_FORMAT (c.updated_at, '%Y-%m-%dT%H:%i:%s') AS updated FROM         result AS c "   commitBatch: 3000 

时间类型的表结构想要存到es中必须自定义转换器或格式化程序，将 Timestamp 转换为 Elasticsearch 支持的日期格式 （例如 ISO 8601 格式）。  否则导入时报错 ERROR c.a.otter.canal.client.adapter.es8x.etl.ESEtlService - cannot write xcontent for unknown value of type class java.sql.Timestamp java.lang.IllegalArgumentException: cannot write xcontent for unknown value of type class java.sql.Timestamp 

因为作者这里有两张表，表结构大同小异

notice 表

result 表

然后启动canal.adapter

赋权： cd /usr/local/canal.adapter/ chmod 777 -R conf/es8 ./bin/startup.sh   tail -f logs/adapter/adapter.log  2024-08-30 15:06:03.275 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Start adapter for canal-client mq topic: example-gl succeed 2024-08-30 15:06:03.275 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## the canal client adapters are running now ...... 2024-08-30 15:06:03.275 [Thread-3] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <============= 2024-08-30 15:06:03.285 [main] INFO  c.a.otter.canal.adapter.launcher.CanalAdapterApplication - Started CanalAdapterApplication in 3.543 seconds (JVM running for 4.264) 2024-08-30 15:06:03.368 [Thread-3] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Subscribe destination: example succeed <============= 2024-08-30 15:09:27.964 [http-nio-8081-exec-1] INFO  o.a.catalina.core.ContainerBase.[Tomcat].[localhost].[/] - Initializing Spring DispatcherServlet 'dispatcherServlet' 输出如上则为成功 

然后测试

# 插入数据，查看一下Canal.adapter是否可以读到数据 登录到mysql中  INSERT INTO notice (id, title, content, created_at, updated_at) VALUES (22, 'New Notice', 'This is a new notice', NOW(), NOW()); INSERT INTO notice (id, title, content, created_at, updated_at) VALUES (23, 'New Notice', 'This is a new notice', NOW(), NOW());  tail -f logs/adapter/adapter.log  输出如下：说明成功 2024-08-30 15:03:05.827 [pool-3-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":22,"title":"New Notice","content":"This is a new notice","created_at":1725001385000,"updated_at":1725001385000}],"database":"info","destination":"example","es":1725001385000,"groupId":"gl","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"notice","ts":1725001385626,"type":"INSERT"} 2024-08-30 15:03:15.858 [pool-3-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":23,"title":"New Notice","content":"This is a new notice","created_at":1725001395000,"updated_at":1725001395000}],"database":"info","destination":"example","es":1725001395000,"groupId":"gl","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"notice","ts":1725001395857,"type":"INSERT"} 

然后全量导入一次数据 curl "localhost:8081/etl/es8/esMappingNotice.yml" -X POST {"succeeded":true,"resultMessage":"导入ES 数据：23 条"}  esMappingNotice.yml 则为适配器文件的名称。  curl "localhost:8081/etl/es8/esMappingResult.yml" -X POST {"succeeded":true,"resultMessage":"导入ES 数据：20 条"} 

然后打开kibana 或者 elasticsearch-head-5.0.0

作者这里使用的是 elasticsearch-head-5.0.0

验证同步配置

在 MySQL 数据库中执行一些增、删、改操作，例如： 登录到mysql中， INSERT INTO notice (id, title, content, created_at, updated_at) VALUES (1, 'New Notice', 'This is a new notice', NOW(), NOW()); UPDATE notice SET content = 'Updated content' WHERE id = 1; DELETE FROM notice WHERE id = 1; 

#### 查询_id 为 10 11 的sql 已经删除 

补充：安装elasticsearch-head

# 安装依赖： yum -y install nodejs  bzip2  # 下载 elasticsearch-head # github地址： https://github.com/mobz/elasticsearch-head # 下载软件包 git clone https://gitee.com/cyberton/elasticsearch-head.git 或者 wget https://github.com/mobz/elasticsearch-head/archive/refs/tags/v5.0.0.tar.gz tar -xf v5.0.0.tar.gz -C /usr/local/ cd /usr/local/elasticsearch-head-5.0.0/  vim Gruntfile.js             connect: {                     server: {                     options: {                        hostname: '*',  # 新增                        port: 9100,                        base: '.',                        keepalive: true                             }                     }             } # 保存退出	  vim  _site/app.js +4388 # 将localhost 修改为 本机IP this.base_uri = this.config.base_uri || this.prefs.get("app-base_uri") || "http://192.168.174.136:9200"; # 保存退出  # 安装依赖 npm install -g grunt-cli # 如果下载失败，更换源 npm cache clean --force npm config set registry https://mirrors.huaweicloud.com/repository/npm/  # 继续安装 npm install -g grunt-cli npm install grunt-contrib-jasmine npm install -g @angular/cli  npm install 有报错先不用处理 直接运行下面命令试试 npm run start   nohup grunt server >/dev/null 2>&1 &   后台运行  访问：稍后测试： http://192.168.174.136:9100 

完工，撒花 🎉🎉🎉🎉🎉

如有问题，请留言评论。

参考文献：

https://blog.csdn.net/qq_30614345/article/details/134083545
https://blog.csdn.net/H_Sino/article/details/137765283