基于Canal与Flink实现数据实时增量同步(一)

这篇具有很好参考价值的文章主要介绍了基于Canal与Flink实现数据实时增量同步(一)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

vi conf/application.yml

server:

port: 8089

spring:

jackson:

date-format: yyyy-MM-dd HH:mm:ss

time-zone: GMT+8

spring.datasource:

address: kms-1:3306

database: canal_manager

username: canal

password: canal

driver-class-name: com.mysql.jdbc.Driver

url: jdbc:mysql:// s p r i n g . d a t a s o u r c e . a d d r e s s / {spring.datasource.address}/ spring.datasource.address/{spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false

hikari:

maximum-pool-size: 30

minimum-idle: 1

canal:

adminUser: admin

adminPasswd: admin

  • 初始化原数据库

mysql -uroot -p

导入初始化SQL

#注:(1)初始化SQL脚本里会默认创建canal_manager的数据库,建议使用root等有超级权限的账号进行初始化

(2)canal_manager.sql默认会在conf目录下

mysql> source /opt/modules/canal-admin/conf/canal_manager.sql

  • 启动canal-admin

sh bin/startup.sh

  • 访问

可以通过 http://kms-1:8089/ 访问,默认密码:admin/123456

  • canal-server端配置

使用canal_local.properties的配置覆盖canal.properties,将下面配置内容配置在canal_local.properties文件里面,就可以了。

register ip

canal.register.ip =

canal admin config

canal.admin.manager = 127.0.0.1:8089

canal.admin.port = 11110

canal.admin.user = admin

canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441

admin auto register

canal.admin.register.auto = true

canal.admin.register.cluster =

  • 启动canal-serve

sh bin/startup.sh local

注意:先启canal-server,然后再启动canal-admin,之后登陆canal-admin就可以添加serve和instance了。

启动kafka控制台消费者测试


bin/kafka-console-consumer.sh --bootstrap-server kms-2:9092,kms-3:9092,kms-4:9092 --topic test --from-beginning

此时MySQL数据表若有变化,会将row类型的log写进Kakfa,具体格式为JSON:

  • insert操作

{

“data”:[

{

“id”:“338”,

“city”:“成都”,

“province”:“四川省”

}

],

“database”:“qfbap_ods”,

“es”:1583394964000,

“id”:2,

“isDdl”:false,

“mysqlType”:{

“id”:“int(11)”,

“city”:“varchar(256)”,

“province”:“varchar(256)”

},

“old”:null,

“pkNames”:[

“id”

],

“sql”:“”,

“sqlType”:{

“id”:4,

“city”:12,

“province”:12

},

“table”:“code_city”,

“ts”:1583394964361,

“type”:“INSERT”

}

  • update操作

{

“data”:[

{

“id”:“338”,

“city”:“绵阳市”,

“province”:“四川省”

}

],

“database”:“qfbap_ods”,

“es”:1583395177000,

“id”:3,

“isDdl”:false,

“mysqlType”:{

“id”:“int(11)”,

“city”:“varchar(256)”,

“province”:“varchar(256)”

},

“old”:[

{

“city”:“成都”

}

],

“pkNames”:[

“id”

],

“sql”:“”,

“sqlType”:{

“id”:4,

“city”:12,

“province”:12

},

“table”:“code_city”,

“ts”:1583395177408,

“type”:“UPDATE”

}

  • delete操作

{

“data”:[

{

“id”:“338”,

“city”:“绵阳市”,

“province”:“四川省”

}

],

“database”:“qfbap_ods”,

“es”:1583395333000,

“id”:4,

“isDdl”:false,

“mysqlType”:{

“id”:“int(11)”,

“city”:“varchar(256)”,

“province”:“varchar(256)”

},

“old”:null,

“pkNames”:[

“id”

],

“sql”:“”,

“sqlType”:{

“id”:4,

“city”:12,

“province”:12

},

“table”:“code_city”,

“ts”:1583395333208,

“type”:“DELETE”

}

JSON日志格式解释

  • data:最新的数据,为JSON数组,如果是插入则表示最新插入的数据,如果是更新,则表示更新后的最新数据,如果是删除,则表示被删除的数据

  • database:数据库名称

  • es:事件时间,13位的时间戳

  • id:事件操作的序列号,1,2,3…

  • isDdl:是否是DDL操作

  • mysqlType:字段类型

  • old:旧数据

  • pkNames:主键名称

  • sql:SQL语句

  • sqlType:是经过canal转换处理的,比如unsigned int会被转化为Long,unsigned long会被转换为BigDecimal

  • table:表名

自我介绍一下,小编13年上海交大毕业,曾经在小公司待过,也去过华为、OPPO等大厂,18年进入阿里一直到现在。

深知大多数Java工程师,想要提升技能,往往是自己摸索成长或者是报班学习,但对于培训机构动则几千的学费,着实压力不小。自己不成体系的自学效果低效又漫长,而且极易碰到天花板技术停滞不前!

因此收集整理了一份《2024年Java开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友,同时减轻大家的负担。基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

既有适合小白学习的零基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!

由于文件比较大,这里只是将部分目录截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且会持续更新!

如果你觉得这些内容对你有帮助,可以扫码获取!!(备注Java获取)

基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

最后

分享一套我整理的面试干货,这份文档结合了我多年的面试官经验,站在面试官的角度来告诉你,面试官提的那些问题他最想听到你给他的回答是什么,分享出来帮助那些对前途感到迷茫的朋友。

面试经验技巧篇
  • 经验技巧1 如何巧妙地回答面试官的问题
  • 经验技巧2 如何回答技术性的问题
  • 经验技巧3 如何回答非技术性问题
  • 经验技巧4 如何回答快速估算类问题
  • 经验技巧5 如何回答算法设计问题
  • 经验技巧6 如何回答系统设计题
  • 经验技巧7 如何解决求职中的时间冲突问题
  • 经验技巧8 如果面试问题曾经遇见过,是否要告知面试官
  • 经验技巧9 在被企业拒绝后是否可以再申请
  • 经验技巧10 如何应对自己不会回答的问题
  • 经验技巧11 如何应对面试官的“激将法”语言
  • 经验技巧12 如何处理与面试官持不同观点这个问题
  • 经验技巧13 什么是职场暗语

基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

面试真题篇
  • 真题详解1 某知名互联网下载服务提供商软件工程师笔试题
  • 真题详解2 某知名社交平台软件工程师笔试题
  • 真题详解3 某知名安全软件服务提供商软件工程师笔试题
  • 真题详解4 某知名互联网金融企业软件工程师笔试题
  • 真题详解5 某知名搜索引擎提供商软件工程师笔试题
  • 真题详解6 某初创公司软件工程师笔试题
  • 真题详解7 某知名游戏软件开发公司软件工程师笔试题
  • 真题详解8 某知名电子商务公司软件工程师笔试题
  • 真题详解9 某顶级生活消费类网站软件工程师笔试题
  • 真题详解10 某知名门户网站软件工程师笔试题
  • 真题详解11 某知名互联网金融企业软件工程师笔试题
  • 真题详解12 国内某知名网络设备提供商软件工程师笔试题
  • 真题详解13 国内某顶级手机制造商软件工程师笔试题
  • 真题详解14 某顶级大数据综合服务提供商软件工程师笔试题
  • 真题详解15 某著名社交类上市公司软件工程师笔试题
  • 真题详解16 某知名互联网公司软件工程师笔试题
  • 真题详解17 某知名网络安全公司校园招聘技术类笔试题
  • 真题详解18 某知名互联网游戏公司校园招聘运维开发岗笔试题

基于Canal与Flink实现数据实时增量同步(一),程序员,flink,linq,数据库

资料整理不易,点个关注再走吧
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》点击传送门即可获取!

  • 真题详解6 某初创公司软件工程师笔试题
  • 真题详解7 某知名游戏软件开发公司软件工程师笔试题
  • 真题详解8 某知名电子商务公司软件工程师笔试题
  • 真题详解9 某顶级生活消费类网站软件工程师笔试题
  • 真题详解10 某知名门户网站软件工程师笔试题
  • 真题详解11 某知名互联网金融企业软件工程师笔试题
  • 真题详解12 国内某知名网络设备提供商软件工程师笔试题
  • 真题详解13 国内某顶级手机制造商软件工程师笔试题
  • 真题详解14 某顶级大数据综合服务提供商软件工程师笔试题
  • 真题详解15 某著名社交类上市公司软件工程师笔试题
  • 真题详解16 某知名互联网公司软件工程师笔试题
  • 真题详解17 某知名网络安全公司校园招聘技术类笔试题
  • 真题详解18 某知名互联网游戏公司校园招聘运维开发岗笔试题

[外链图片转存中…(img-SMAMbiDo-1712418305578)]

资料整理不易,点个关注再走吧
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》点击传送门即可获取!文章来源地址https://www.toymoban.com/news/detail-849755.html

到了这里,关于基于Canal与Flink实现数据实时增量同步(一)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于Canal实现MySQL 8.0 数据库数据同步

    主机名称 操作系统 说明 192.168.11.82 Ubuntu 22.04 主库所在服务器 192.168.11.28 Oracle Linux Server 8.7 从库所在服务器 1、Ubuntu系统下MySQL配置文件位置 2、CentOS系统下MySQL配置文件位置 3、添加如下配置,开启MySQL binlog功能 关于canal简介,这里就不再阐述,具体可以参看官方文档介绍,地

    2023年04月23日
    浏览(196)
  • Canal实时同步MySQL数据到ES

    canal主要用途是对MySQL数据库增量日志进行解析,提供增量数据的订阅和消费,简单说就是可以对MySQL的增量数据进行实时同步,支持同步到MySQL、Elasticsearch、HBase等数据存储中去。 早期阿里巴巴因为杭州和美国双机房部署,存在跨机房同步的业务需求,实现方式主要是基于业

    2024年02月04日
    浏览(61)
  • 基于Flink CDC实时同步数据(MySQL到MySQL)

    jdk8 Flink 1.16.1(部署在远程服务器:192.168.137.99) Flink CDC 2.3.0 MySQL 8.0(安装在本地:192.168.3.31) (安装部署过程略) 准备三个数据库:flink_source、flink_sink、flink_sink_second。 将flink_source.source_test表实时同步到flink_sink和flink_sink_second的sink_test表。 (建库建表过程略) 开发过程

    2024年02月06日
    浏览(104)
  • 基于 Flink CDC 构建 MySQL 到 Databend 的 实时数据同步

    这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 到 Databend 的实时数据同步。本教程的演示都将在 Flink SQL CLI 中进行,只涉及 SQL,无需一行 Java/Scala 代码,也无需安装 IDE。 假设我们有电子商务业务,商品的数据存储在 MySQL ,我们需要实时把它同步到 Databend 中。 接下来的内容

    2024年02月10日
    浏览(53)
  • canal实时同步mysql数据到elasticsearch(部署,配置,测试)(一)

    canal基于MySQL数据库增量日志解析,提供增量数据订阅和消费,是阿里开源CDC工具,它可以获取MySQL binlog数据并解析,然后将数据变动传输给下游。基于canal,可以实现从MySQL到其他数据库的实时同步 MySQL主备复制原理 MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫

    2023年04月08日
    浏览(44)
  • 使用 Docker 部署 canal 服务实现MySQL和ES实时同步

    参考 ClientAdapter: Canal的Adapter配置项目 Sync ES:Canal的Adapter中ES同步的配置项 使用 Docker 部署 canal 服务 docker canal-server canal-adapter mysql Canal(基于Docker同步mysql数据到elasticsearch) Canal部署过程中的错误 Canal 1.1.4 Canal Adapter 1.1.4 Kibana: 6.8.8 ElasticSearch: 6.4.3 由于Canal 1.1.4只能适配 Ela

    2024年02月13日
    浏览(45)
  • MySQL如何实时同步数据到ES?试试阿里开源的Canal

    前几天在网上冲浪的时候发现了一个比较成熟的开源中间件——  Canal  。在了解了它的工作原理和使用场景后,顿时产生了浓厚的兴趣。今天,就让我们跟随我的脚步,一起来揭开它神秘的面纱吧。 目录 前言 简介  工作原理  MySQL主备复制原理 canal 工作原理 Canal架构  C

    2024年02月20日
    浏览(43)
  • [大数据 Flink,Java实现不同数据库实时数据同步过程]

    目录 🌮前言: 🌮实现Mysql同步Es的过程包括以下步骤: 🌮配置Mysql数据库连接 🌮在Flink的配置文件中,添加Mysql数据库的连接信息。可以在flink-conf.yaml文件中添加如下配置: 🌮在Flink程序中,使用JDBCInputFormat来连接Mysql数据库,并定义查询语句,获取需要同步的数据。具体代

    2024年02月10日
    浏览(45)
  • docker环境安装mysql、canal、elasticsearch,基于binlog利用canal实现mysql的数据同步到elasticsearch中

    🚀 本文提供的指令完全可以按顺序逐一执行,已进行了多次测试。因此如果你是直接按照我本文写的指令一条条执行的,而非自定义修改过,执行应当是没有任何问题的。 🚀 本文讲述:使用docker环境安装mysql、canal、elasticsearch,基于binlog利用canal实现mysql的数据同步到elas

    2024年02月02日
    浏览(58)
  • Springcloud Alibaba 使用Canal将MySql数据实时同步到Elasticsearch

    本篇文章在Springcloud Alibaba使用Canal将Mysql数据实时同步到Redis保证缓存的一致性-CSDN博客 基础上使用canal将mysql数据实时同步到Elasticsearch。 公共包 实体类Sku @Column注解 用来标识实体类中属性与数据表中字段的对应关系 name 定义了被标注字段在数据库表中所对应字段的名称;由

    2024年02月03日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包