Canal Mysql binlog 同步至 ElasticSearch 详细介绍

文章目录

  • 数据同步ElasticSearch
    • 单表基本配置
    • 适配器映射文件详细介绍（单表、多表映射介绍）
      • 单表映射索引示例sql
      • 单表映射索引示例sql带函数或运算操作
      • 多表映射(一对一, 多对一)索引示例sql
      • 多表映射(一对多)索引示例sql
      • 其它类型的sql示例
      • 注意事项

本文详细介绍Canal 配置保存 ElasticSearch

Canal从零配置使用参考：https://blog.csdn.net/zhangshenghang/article/details/120361721

数据同步ElasticSearch

我们接着在之前配置Hbase基础上直接修改配置，实现同时同步ElasticSearch

单表基本配置

  • 1.修改启动器配置 {canal-apapter}/conf/application.yml
server:port: 8081
logging:level:com.alibaba.otter.canal.client.adapter: DEBUGcom.alibaba.otter.canal.client.adapter.hbase: DEBUG
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8default-property-inclusion: non_null
canal.conf:# tcp kafka rocketMQ rabbitMQ canal-server运行的模式，TCP模式就是直连客户端，不经过中间件。kafka和mq是消息队列的模式mode: tcp 
#  flatMessage: truezookeeperHosts: syncBatchSize: 1retries: 0timeout: 1000accessKey:secretKey:consumerProperties:# canal tcp consumer 指定canal-server的地址和端口canal.tcp.server.host: 127.0.0.1:11111canal.tcp.zookeeper.hosts: 127.0.0.1:2181canal.tcp.batch.size: 1canal.tcp.username:canal.tcp.password:srcDataSources: # 数据源配置，从哪里获取数据defaultDS: # 指定一个名字，在ES的配置中会用到，唯一url: jdbc:mysql://127.0.0.1:3306/test2?useUnicode=trueusername: rootpassword: *****canalAdapters:- instance: example # canal instance Name or mq topic name 指定在canal配置的实例名称groups:- groupId: g1 outerAdapters:- name: logger
#      - name: rdb
#        key: mysql1
#        properties:
#          jdbc.driverClassName: com.mysql.jdbc.Driver
#          jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
#          jdbc.username: root
#          jdbc.password: 121212
#      - name: rdb
#        key: oracle1
#        properties:
#          jdbc.driverClassName: oracle.jdbc.OracleDriver
#          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
#          jdbc.username: mytest
#          jdbc.password: m121212
#      - name: rdb
#        key: postgres1
#        properties:
#          jdbc.driverClassName: org.postgresql.Driver
#          jdbc.url: jdbc:postgresql://localhost:5432/postgres
#          jdbc.username: postgres
#          jdbc.password: 121212
#          threads: 1
#          commitSize: 3000- name: hbaseproperties:hbase.zookeeper.quorum: sangfor.abdi.node3,sangfor.abdi.node2,sangfor.abdi.node1hbase.zookeeper.property.clientPort: 2181zookeeper.znode.parent: /hbase-unsecure- name: es7 # config目录下的子目录名称hosts: 192.168.168.2:9300 # 127.0.0.1:9200 for rest modeproperties:mode: transport # or rest
#          # security.auth: test:123456 #  only used for rest modecluster.name: my_application
#        - name: kudu
#          key: kudu
#          properties:
#            kudu.master.address: 127.0.0.1 # ',' split multi address
  • 2.ElasticSearch 表映射文件
# 指定数据源，这个值和adapter的application.yml文件中配置的srcDataSources值对应。
dataSourceKey: defaultDS
# 指定canal-server中配置的某个实例的名字，不同实例对应不同业务
destination: example
# 组ID ，tcp方式这里填写空，不要填写值，不然可能会接收不到数据
groupId: 
# ES的mapping（映射）
esMapping:# ES索引名称_index: testsync2# ES标示文档的唯一标示，通常对应数据表中的主键ID字段_id: _id
#  upsert: true
#  pk: id
# 数据表每个字段映射到表中的具体名称，不能重复sql: "select a.id as _id, a.name,a.age,a.age_2,a.message,a.insert_time from testsync as a"
#  objFields:
#    _labels: array:;
#  etlCondition: "where a.c_time>={}"commitBatch: 10
  • 3 重启服务
bin/restart.sh

写入数据

INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());

查看adapter日志

2021-09-20 13:53:07.279 [pool-1-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"}
2021-09-20 13:53:07.286 [pool-1-thread-1] DEBUG c.a.o.c.client.adapter.hbase.service.HbaseSyncService - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"}
2021-09-20 13:53:07.287 [pool-1-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"} 
Affected indexes: testsync2

查看ElasticSearch数据
在这里插入图片描述
至此写入ElasticSearch、Hbase成功

适配器映射文件详细介绍（单表、多表映射介绍）

${adapter}/conf/es7/xxx.yml

dataSourceKey: defaultDS        # 源数据源的key, 对应上面配置的srcDataSources中的值
outerAdapterKey: exampleKey     # 对应application.yml中es配置的key 
destination: example            # cannal的instance或者MQ的topic
groupId:                        # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:_index: mytest_user           # es 的索引名称_type: _doc                   # es 的type名称, es7下无需配置此项_id: _id                      # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
#  pk: id                       # 如果不需要_id, 则需要指定一个属性为主键属性# sql映射sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,a.c_time as _c_time, c.labels as _labels from user aleft join role b on b.id=a.role_idleft join (select user_id, group_concat(label order by id desc separator ';') as labels from labelgroup by user_id) c on c.user_id=a.id"
#  objFields:
#    _labels: array:;           # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的
#    _obj: object               # json对象etlCondition: "where a.c_time>='{0}'"     # etl 的条件参数commitBatch: 3000                         # 提交批大小

sql映射说明:

sql支持多表关联自由组合, 但是有一定的限制:

  1. 主表不能为子查询语句
  2. 只能使用left outer join即最左表一定要是主表
  3. 关联从表如果是子查询不能有多张表
  4. 主sql中不能有where查询条件(从表子查询中可以有where条件但是不推荐, 可能会造成数据同步的不一致, 比如修改了where条件中的字段内容)
  5. 关联条件只允许主外键的’='操作不能出现其他常量判断比如: on a.role_id=b.id and b.statues=1
  6. 关联条件必须要有一个字段出现在主查询语句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主select语句中

Elastic Search的mapping 属性与sql的查询值将一一对应(不支持 select *), 比如: select a.id as _id, a.name, a.email as _email from user, 其中name将映射到es mapping的name field, _email将 映射到mapping的_email field, 这里以别名(如果有别名)作为最终的映射字段. 这里的_id可以填写到配置文件的 _id: _id映射.

单表映射索引示例sql

select a.id as _id, a.name, a.role_id, a.c_time from user a

该sql对应的es mapping示例:

{"mytest_user": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"c_time": {"type": "date"}}}}}
}

单表映射索引示例sql带函数或运算操作

select a.id as _id, concat(a.name,'_test') as name, a.role_id+10000 as role_id, a.c_time from user a

函数字段后必须跟上别名, 该sql对应的es mapping示例:

{"mytest_user": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"c_time": {"type": "date"}}}}}
}

多表映射(一对一, 多对一)索引示例sql

select a.id as _id, a.name, a.role_id, b.role_name, a.c_time from user a 
left join role b on b.id = a.role_id

注:这里join操作只能是left outer join, 第一张表必须为主表!!
该sql对应的es mapping示例:

{"mytest_user": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"role_name": {"type": "text"},"c_time": {"type": "date"}}}}}
}

多表映射(一对多)索引示例sql

select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a 
left join (select user_id, group_concat(label order by id desc separator ';') as labels from labelgroup by user_id) c on c.user_id=a.id

注:left join 后的子查询只允许一张表, 即子查询中不能再包含子查询或者关联!!

该sql对应的es mapping示例:

{"mytest_user": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"c_time": {"type": "date"},"labels": {"type": "text"}}}}}
}

其它类型的sql示例

  • geo type
select ... concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location, ...
  • 复合主键
select concat(a.id,'_',b.type) as _id, ... from user a left join role b on b.id=a.role_id
  • 数组字段
select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a 
left join (select user_id, group_concat(label order by id desc separator ';') as labels from labelgroup by user_id) c on c.user_id=a.id

配置中使用:

objFields:labels: array:;
  • 对象字段
select a.id as _id, a.name, a.role_id, c.labels, a.c_time, a.description from user a

配置中使用:

objFields:description: object

其中a.description字段内容为json字符串

  • 父子文档索引
    es/customer.yml
......
esMapping:_index: customer_type: _doc_id: idrelations:customer_order:name: customersql: "select t.id, t.name, t.email from customer t"

es/order.yml

esMapping:_index: customer_type: _doc_id: _idrelations:customer_order:name: orderparent: customer_idsql: "select concat('oid_', t.id) as _id,t.customer_id,t.id as order_id,t.serial_code as order_serial,t.c_time as order_timefrom biz_order t"skips:- customer_id

mapping示例:

{"mappings":{"_doc":{"properties":{"id": {"type": "long"},"name": {"type": "text"},"email": {"type": "text"},"order_id": {"type": "long"},"order_serial": {"type": "text"},"order_time": {"type": "date"},"customer_order":{"type":"join","relations":{"customer":"order"}}}}}
}

注意事项

  • 多表映射时，主表数据必须插入，如果只插入子表不插入主表，数据无法同步到ElasticSearch；相反只插入主表，子表不进行插入，数据是可以同步到ElasticSearch的
  • 多表映射时，如果主表关联id写入后，子表再进行修改之前的关联的id为我们主表写入的id，数据是无法同步到ElasticSearch中的。
    在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/509627.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

基于C++11的线程池

基于C++11的线程池

背景 在传统的收到任务即创建线程的情况下，我们每收到一个任务，就创建一个线程，执行任务，销毁线程， 我们把这三个过程所用的时间分别记做T1,T2,T3 任务本身所用的时间仅占T2/(T1T2T3),这在任务本身所用时间很短的情况下…
阅读更多...
集合的工具类

集合的工具类

集合操作的工具类: 1):Arrays类: 2):Collections类. Arrays类: 在Collection接口中有一个方法叫toArray把集合转换为Object数组. 把集合转换为数组: Object[] arr 集合对象.toArray(); 数组也可以转换为集合(List集合): public static List asList(T… a) 等价于public …
阅读更多...
Docker入门到精通开发指南(一文搞懂)

Docker入门到精通开发指南(一文搞懂)

文章目录安装官方安装文档具体安装步骤1.卸载之前的版本(如果之前未用过忽略该步骤)2.安装相关依赖3.设置docker镜像4.安装docker安装latest版本指定版本安装5.启动docker6.查看docker版本7.运行一个docker hello world8.卸载docker9.配置阿里云镜像加速地址docker常用命令dock…
阅读更多...
数据结构实验之串二：字符串匹配

数据结构实验之串二：字符串匹配

题目描述 给定两个字符串string1和string2，判断string2是否为string1的子串。 输入 输入包含多组数据，每组测试数据包含两行，第一行代表string1，第二行代表string2，string1和string2中保证不出现空格。(string1和strin…
阅读更多...
Unicode和Utf-8

Unicode和Utf-8

很久很久以前，有一群人，他们决定用8个可以开合的晶体管来组合成不同的状态，以表示世界上的万物。他们看到8个开关状态是好的，于是他们把这称为”字节“。再后来，他们又做了一些可以处理这些字节的机器，机器…
阅读更多...
java集合框架总结之思维导图

java集合框架总结之思维导图

思维导图
阅读更多...
如何向Maven中央仓库提交自己的Jar包(发布自己的Jar包到中央仓库)

如何向Maven中央仓库提交自己的Jar包(发布自己的Jar包到中央仓库)

文章目录注册账号GPG 安装安装生成密钥上传公钥Maven配置上传到Maven仓库修改项目的配置，填写基本信息执行编译命令登录网站配置发布项目中应用遇到的问题解决方法本文将介绍如何将自己的jar包发布至公共的中央仓库，通过maven方式进行引用 注册账号 注册…
阅读更多...
数据结构实验之串三：KMP应用

数据结构实验之串三：KMP应用

题目描述 有n个小朋友&#xff0c;每个小朋友手里有一些糖块&#xff0c;现在这些小朋友排成一排&#xff0c;编号是由1到n。现在给出m个数&#xff0c;能不能唯一的确定一对值l和r(l < r)&#xff0c;使得这m个数刚好是第l个小朋友到第r个小朋友手里的糖块数&#xff1f;输…
阅读更多...
C++读写锁

C++读写锁

读写锁实际是一种特殊的自旋锁&#xff0c;它把对共享资源的访问者划分成读者和写者&#xff0c;读者只对共享资源进行读访问&#xff0c;写者则需要对共享资源进行写操作。 读写锁实际是一种特殊的自旋锁&#xff0c;它把对共享资源的访问者划分成读者和写者&#xff0c;读者只…
阅读更多...
List和Set以及Map的选用

List和Set以及Map的选用

选用哪一种容器取决于每一种容器的存储特点以及当前业务的需求: List: 单一元素集合. 允许元素重复/记录元素的添加顺序. Set:单一元素集合. 不允许元素重复/不记录元素的添加顺序. 既要不重复,又要保证先后顺序:LinkedHashSet. Map: 双元素集合. 如果存储数据的时候,还得…
阅读更多...
FastDFS文件服务器安装

FastDFS文件服务器安装

文章目录环境准备使用的系统软件磁盘目录安装libfastcommon安装FastDFS安装fastdfs-nginx-module安装nginx单机部署tracker配置storage配置client测试配置nginx访问分布式部署tracker配置storage配置client测试配置nginx访问启动防火墙trackerstoragenginx检测集群说明配置文件…
阅读更多...
学密码学一定得学程序

学密码学一定得学程序

题目描述 曾经&#xff0c;ZYJ同学非常喜欢密码学。有一天&#xff0c;他发现了一个很长很长的字符串S1。他很好奇那代表着什么&#xff0c;于是神奇的WL给了他另一个字符串S2。但是很不幸的是&#xff0c;WL忘记跟他说是什么意思了。这个时候&#xff0c;ZYJ不得不求助与伟大的…
阅读更多...
boost::shared_mutex

boost::shared_mutex

前一篇介绍的条件变量可以进行进程间通信&#xff0c;用来实现生产者/消费者模型。今天介绍的共享互斥量用来实现缓冲区读写模型&#xff0c;与生产者/消费者模型不同的地方是&#xff0c;消费者消费产品后&#xff0c;被消费的产品就不存在了&#xff0c;所以消费者线程也要互…
阅读更多...
Map集合类

Map集合类

映射的数学解释: 设A、B是两个非空集合&#xff0c;如果存在一个法则f&#xff0c;使得对A中的每个元素a&#xff0c;按法则f&#xff0c;在B中有唯一确定的元素b与之对应&#xff0c;则称f为从A到B的映射&#xff0c;记作f&#xff1a;A→B。 映射关系(两个集合):A集合和B集…
阅读更多...
github 国内加速镜像

github 国内加速镜像

GitHub 镜像访问 这里提供两个最常用的镜像地址&#xff08;别登录账号&#xff09;&#xff1a; https://github.com.cnpmjs.org https://hub.fastgit.org 也就是说上面的镜像就是一个克隆版的Github&#xff0c;你可以访问上面的镜像网站&#xff0c;网站的内容跟Github是完…
阅读更多...
数据结构实验之二叉树二：遍历二叉树

数据结构实验之二叉树二：遍历二叉树

题目描述 已知二叉树的一个按先序遍历输入的字符序列&#xff0c;如abc,,de,g,,f,,, (其中,表示空结点)。请建立二叉树并按中序和后序的方式遍历该二叉树。 输入 连续输入多组数据&#xff0c;每组数据输入一个长度小于50个字符的字符串。 输出 每组输入数据对应输出2行&#x…
阅读更多...
Socket select模型

Socket select模型

Windows socket select模型开发。 套接字select模型是一种比较常用的IO模型。利用该模型可以使Windows socket应用程序可以同时管理多个套接字。 使用select模型&#xff0c;可以使当执行操作的套接字满足可读可写条件时&#xff0c;给应用程序发送通知。收到这个通知后&#x…
阅读更多...
Set实现类性能对比

Set实现类性能对比

Set接口的实现类: 共同的特点: 1):都不允许元素重复. 2):都不是线程安全的类. 解决方案:Set s Collections.synchronizedSet(Set对象); HashSet: 不保证元素的先后添加顺序. 底层才有的是哈希表算法,查询效率极高. 判断两个对象是否相等的规则: 1):equals比较为true. …
阅读更多...
RSocket协议初识

RSocket协议初识

文章目录前言RSocket是什么&#xff1f;RSocket设计目标是什么&#xff1f;RSocket与其他协议有什么区别&#xff1f;对比Http1.x对比Http2.x对比grpc对比TCP对比WebSocket结论RSocket适用于哪些场景&#xff1f;1、移动设备与服务器的连接。2、微服务场景。3、由于微服务和移动…
阅读更多...
HugeGraph Server/Hubble安装使用

HugeGraph Server/Hubble安装使用

文章目录HugeGraph Server1 概述2 依赖2.1 安装JDK-1.83 部署3.1 下载tar包4 安装启动4.1 解压4.2 配置Hbase5 访问Server5.1 服务启动状态校验6 停止Server7 多图配置HugeGraph-Hubble 基于Web的可视化图形界面1.概述2.安装3 使用3.1创建图HugeGraph Server 1 概述 HugeGrap…
阅读更多...
最新文章

Copyright @ 2022~2023