实现MySQL同步数据到ES构建宽表

article/2025/10/12 13:09:52

作者介绍

Ceven，德勤乐融(北京)科技有限公司 邮箱：likailin@deqinyuerong.com

前言

CloudCanal 近期提供了自定义代码构建宽表能力，我们第一时间参与了该特性内测，效果不错。开发流程详见官方文档 《CloudCanal自定义代码实时加工》

能力特点包括：

  • 灵活，支持反查打宽表，特定逻辑数据清洗，对账，告警等场景
  • 调试方便，通过任务参数配置自动打开 debug 端口，对接 IDE 调试
  • SDK 接口清晰，提供丰富的上下文信息，方便数据逻辑开发

本文基于我们业务中的实际需求(MySQL -> ElasticSearch 宽表构建)，梳理一下具体的开发调试流程，希望对大家有所帮助。

场景描述

MySQL 擅长关系型数据操作，我们在其中存储了 product, tag, product_tag_mapping 表数据，用以表示产品和标签之间多对多关系。精简的数据结构如下：

88ae6c35-4519-4d51-b725-d05765d67b06-image.png

ElasticSearch 擅长搜索，但是并不支持不同索引间的联合查询, 所以构造宽表是业界刚需。我们存储其上的产品索引结构如下：

PUT es_product
{"mappings" : {"properties" : {"id" : {"type" : "integer"},"name" : {"type" : "text"},"tags" : {"type" : "nested", "properties" : {"id" : {"type" : "integer"},"name" : {"type" : "text"}}}}}
}

同步策略

CloudCanal 在 同步 MySQL -> ElasticSearch 数据过程中，会兼顾全量和增量两种情况，我们可以创建两个独立的任务，分别同步产品的基础信息和附加信息（即标签信息）。

  • 基础信息任务
    • 使用基本的映射关系，将 MySQL 中的 product 数据表，映射到 es_product 索引中，即可保证全量和增量的数据同步。
  • 附加信息任务
    • 创建 CloudCanal 任务将 MySQL 中的 product_tag_mapping 数据表映射到 es_product 索引中，同步过程中反查源数据库中的 tag 信息，构造宽表数据，填充进 es_product 索引，实现附加信息全量和增量的数据同步。

实现步骤

1. MySQL 表结构初始化

# 创建产品信息表
CREATE TABLE `product` (`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,`name` varchar(64) COLLATE utf8_unicode_ci NOT NULL DEFAULT '' COMMENT '名称',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='产品信息记录表';# 创建标签信息表
CREATE TABLE `tag` (`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,`name` varchar(64) COLLATE utf8_unicode_ci NOT NULL DEFAULT '' COMMENT '名称',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='标签信息记录表';# 创建产品标签关系表
CREATE TABLE `product_tag_mapping` (`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,`product_id` bigint(20) unsigned NOT NULL DEFAULT '0' COMMENT '产品ID',`tag_id` bigint(20) unsigned NOT NULL DEFAULT '0' COMMENT '标签ID',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='产品标签关系表';

2. MySQL 填充测试数据

# 填充产品信息
INSERT INTO `product` (`name`)
VALUES('product_1');# 填充标签信息
INSERT INTO `tag` (`name`)
VALUES('tag_1'),('tag_2');# 填充产品标签关系信息
INSERT INTO `product_tag_mapping` (`product_id`, `tag_id`)
VALUES(1, 1);

3. ElasticSearch 索引创建（也可以使用 CloudCanal 结构迁移）

PUT es_product
{"mappings" : {"properties" : {"id" : {"type" : "integer"},"name" : {"type" : "text"},"tags" : {"type" : "nested", "properties" : {"id" : {"type" : "integer"},"name" : {"type" : "text"}}}}}
}

4. 编写自定义代码

自定义代码的项目基于 maven 构建，可以参考 示例项目 cloudcanal-sdk-demos

4.1 修改 MAVEN 配置

初始化的项目需要手工配置一下 pom.xml 文件，将 sdk 指向本地目录文件，代码片段如下

<dependency><groupId>com.clougence.cloudcanal</groupId><artifactId>cloudcanal-sdk</artifactId><version>1.0.0-SNAPSHOT</version><scope>system</scope><systemPath>/path/to/your/project/src/main/resources/lib/cloudcanal-sdk-2.0.0.9-SNAPSHOT.jar</systemPath>
</dependency>

4.2 实现 TAG 类

public class Tag {private int id;private String name;public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}
}

4.3 实现 PROCESSOR 处理逻辑

        @Overridepublic List<CustomRecord> process(List<CustomRecord> list, CustomProcessorContext context) {DataSource dataSource = (DataSource) context.getProcessorContextMap().get(RdbContextKey.SOURCE_DATASOURCE);String stage = context.getProcessorContextMap().get("currentTaskStage").toString();for (CustomRecord record : list) {try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement()) {// 由于 ES 的嵌套结构会被认为是独立的文档，故需要填充旧的数据ResultSet rs = statement.executeQuery("SELECT `tag`.`id`, `tag`.`name`" +" FROM `product`.`product_tag_mapping` AS `mapping`" +" LEFT JOIN `product`.`tag` AS `tag` ON `tag`.`id` = `mapping`.`tag_id`" +" WHERE `mapping`.`product_id` = " + record.getFieldMapAfter().get("product_id").getValue());List<Tag> tags = buildTags(rs);if ("INCREMENT".equals(stage)) {// 增量创建的 product_tag_mapping 处于内存中，无法通过 SQL 语句查询得到，故需要单独处理rs = statement.executeQuery("SELECT `id`, `name` FROM `product`.`tag` WHERE `id` = " + record.getFieldMapAfter().get("tag_id").getValue().toString());List<Tag> newTags = buildTags(rs);tags.add(newTags.get(0));}ObjectMapper mapper = new ObjectMapper();String json = mapper.writeValueAsString(tags);Map<String, Object> tagField = new LinkedHashMap<>();tagField.put("tags", json);RecordBuilder.modifyRecordBuilder(record).addField(tagField).build();} catch (SQLException | JsonProcessingException e) {e.printStackTrace();}}return list;}private List<Tag> buildTags(ResultSet rs) throws SQLException {List<Tag> tags = new ArrayList<>();while (rs.next()) {Tag tag = new Tag();tag.setId(rs.getInt("id"));tag.setName(rs.getString("name"));tags.add(tag);}return tags;}

4.4 编译自定义代码包

执行如下命令编译生成自定义代码包, 之后会在 target 目录中生成 jar 文件

mvn clean package -Dmaven.test.skip=true -Dmaven.compile.fork=true

5. 创建 CloudCanal 任务

5.1 同步 PRODUCT 基础数据

全量增量同步 product 信息到 es_product 索引，在此就不做具体描述，详情请参考 CloudCanal 文档。

此时查询产品数据，得到结果

787f8ce4-6ad8-4d57-8a05-5694c705fed1-image.png

5.2 扩展 PRODUCT TAG 数据

5.2.1 配置数据源和目标

b8b1f5ec-3e3c-4620-ba87-ba224ca265e1-image.png

5.2.2 配置规格

可去掉自动启动任务选项，以便于单步追踪调试 8b1e059d-b4cb-4795-b27e-50cb5ae2c2a3-image.png

5.2.3 配置索引映射

Tips: 只配置增加操作，不要配置编辑和删除，否则可能造成对数据的误删；编辑和删除操作，只最好使用 ES 调用的方式进行处理；增加操作最好不要使用 ES 调用的方式处理，会引起高并发问题。

5.2.4 上传自定义代码

4b8abc11-5c10-4be5-932a-b4dfc6e7740f-image.png

f1e71074-7ce3-48ec-a162-b1814fe928bb-image.png

Tips: 创建任务时如果不上传自定义代码包，之后将无法上传，除非重建任务。上传自定义代码，意味着创建特殊类型的任务，然后才会出现特殊的选项进行字段映射。

5.2.5 配置字段映射

将 id 和 tag_id 调整为 “只订阅不同步”(老版本此处会显示为仅供自定义代码使用)，实现只订阅这两个字段，而不会真正写入到 ES 索引，而将 product_id 映射到对端的 id。 1ec04979-b240-4953-8026-dbecbde0c886-image.png

设置映射 _id，以指定目标 ES 索引中的 id 为 product_id

513633e9-a603-43d5-b9f1-6d6b7b0cd504-image.png

b1419349-20cc-4c4d-a09b-a75bc7a9218b-image.png

Tips: product_id 字段必须做映射，否则即使配置了 _id 信息，依旧无法正常执行，会忽略 product_id 字段的值。

6. 同步结果

87ec9e06-17ac-4bed-b307-79e17cca03ea-image.png

调试自定义代码

自定义代码在开发阶段最麻烦的事情是如何高效进行调试，CloudCanal 能够比较友好的让开发在本地直接调试代码逻辑。

修改任务参数

任务详情->参数修改

f3f59272-9b6a-40f9-ac3e-618782833676-image.png

00072b56-dbe0-4ce0-939a-7e22141419d5-image.png

Tips：每次修改完参数信息之后，必须点击生效配置和重启任务；在任务详情配置中，也可以上传新的代码包，激活和重启任务后可以使用。

配置 IntelliJ IDEA Debug 模式

b29b139e-1ffb-409c-bad5-6ee7ae76863b-image.png

Tips: 设置好断点以后，需要先启动 CloudCanal 任务，再点击 debug 按钮，才能 Attach 到远程的 8787 端口；CloudCanal 会一直 pending，直到有 Attachment，才会继续执行，所以不需要单步跟踪调试时，一定记得关闭调试模式，否则任务无法执行。

总结

CloudCanal 自定义代码能够拓展的能力具有不错的想象空间，我们甚至能加入一些在线业务逻辑的处理，让业务需求能够更好的满足，同时配合社区版调试也很方便。希望未来这块能力在便利功能，性能等层面有更好的表现。

参与内测

CloudCanal 会不断提供一些预览的能力，包括新数据链路, 优化能力,功能插件。本文所描述的自定义代码能力目前也处于内测阶段。如需体验，可添加我们小助手(微信号:suhuayue001)进行了解和试用。

加入CloudCanal粉丝群掌握一手消息和获取更多福利，请添加我们小助手微信：suhuayue001

CloudCanal-免费好用的企业级数据同步工具，欢迎品鉴。 了解更多产品可以查看官方网站： http://www.clougence.com CloudCanal社区：https://www.askcug.com/


http://chatgpt.dhexx.cn/article/IgRKdT6Y.shtml

相关文章

宽表：数据仓库 - “宽表”之争？

宽表：数据仓库 - “宽表”之争？

昨天在技术交流群里一个问题引发了激烈的讨论&#xff0c;我决定把它记录下来。 问题如下&#xff1a;DWD 中有宽表么&#xff1f; 作为扫盲文章&#xff0c;基础知识我们再普及一下&#xff0c;先介绍下基础相关概念。 数仓分层&#xff08;来自&#xff1a;个人理解&#xff…
阅读更多...
Elasticseach：从微服务架构演变到大宽表思维的架构转变

Elasticseach：从微服务架构演变到大宽表思维的架构转变

序言 图示&#xff1a;Elasticsearch 在DB-Engine综合排名第8 Elasticsearch 简称"ES”, 在DB-Engine 综合排名第8&#xff0c;已经持续了相当长的时间&#xff0c;按照当下热度应该会继续保持或者上升一个名次&#xff1b;ES在多数工程师印象中最深刻可能是ELK三件套或者…
阅读更多...
9.Flink实时项目之订单宽表

9.Flink实时项目之订单宽表

1.需求分析 订单是统计分析的重要的对象&#xff0c;围绕订单有很多的维度统计需求&#xff0c;比如用户、地区、商品、品类、品牌等等。为了之后统计计算更加方便&#xff0c;减少大表之间的关联&#xff0c;所以在实时计算过程中将围绕订单的相关数据整合成为一张订单的宽表…
阅读更多...
数仓建模—宽表的设计

数仓建模—宽表的设计

宽表的设计 高内聚低耦合 宽表是数仓里面非常重要的一块&#xff0c;数仓是分层的&#xff0c;这是技术进步和时代变化相结合的产物&#xff0c;数仓的分层式为了更好地管理数仓以及更加高效地进行数据开发。 宽表主要出现在dwd 层和报表层&#xff0c;当然有的人说dws 层也有…
阅读更多...
基于宽表的数据建模应用

基于宽表的数据建模应用

一、业务背景 1.1 数据建模现状 互联网企业往往存在多个产品线&#xff0c;每天源源不断产出大量数据&#xff0c;这些数据服务于数据分析师、业务上的产品经理、运营、数据开发人员等各角色。为了满足这些角色的各种需求&#xff0c;业界传统数仓常采用的是经典分层模型的数…
阅读更多...
数据仓库宽表

数据仓库宽表

1. 构建宽表的目的 讲宽表我想从为什么需要宽表入手&#xff0c;而不是一上来就抠概念。因为我觉得一门知识叫什么名字并不是最核心的&#xff0c;关键是搞清楚它的诞生背景以及如何在特定场景用好它。 构建宽表的目的很简单,就是为了"一站式"尽可能多的展示我们需要…
阅读更多...
宽表, 窄表, 维度表, 事实表的区别

宽表, 窄表, 维度表, 事实表的区别

在数据开发里, 会涉及到一些概念: 宽表, 窄表, 维度表, 事实表 宽表: 把多个维度的字段都放在一张表存储, 增加数据冗余是为了减少关联, 便于查询. 查询一张表就可以查出不同维度的多个字段窄表: 和我们 mysql 普通表三范式相同, 把相同维度的字段组成一张表, 表和表之间关联查…
阅读更多...
[转]科普 | 什么是宽表？

[转]科普 | 什么是宽表？

科普 | 什么是宽表&#xff1f;一文带你了解 数据仓库宽表_数据宽表_吕归尘0的博客-CSDN博客 一、什么是“宽表”&#xff1f; “宽表”从字面上的意思就是字段&#xff08;列&#xff09;比较多的数据库表&#xff0c;是通过关联字段将多个业务主题相关的数据表进行挂接组装…
阅读更多...
数仓建模，宽表是什么？如何设计？

数仓建模，宽表是什么？如何设计？

数仓建模&#xff0c;宽表是什么&#xff1f;如何设计&#xff1f; 宽表的设计为什么要建设宽表宽表的好处和不足如何设计宽表总结 宽表的设计 其实宽表是数仓里面非常重要的一块&#xff0c;宽表主要出现在dwd 层和报表层&#xff0c;当然有的人说dws 层也有宽表&#xff0c;…
阅读更多...
线性代数笔记22——特征值和特征向量

线性代数笔记22——特征值和特征向量

特征向量 函数通常作用在数字上&#xff0c;比如函数f作用在x上&#xff0c;结果得到了f(x)。在线性代数中&#xff0c;我们将x扩展到多维&#xff0c;对于Ax来说&#xff0c;矩阵A的作用就像一个函数&#xff0c;输入一个向量x&#xff0c;通过A的作用&#xff0c;得到向量Ax。…
阅读更多...
特征值和特征向量意义

特征值和特征向量意义

本文转载自https://blog.csdn.net/fuming2021118535/article/details/51339881&#xff0c;版权问题请联系博主删除 在刚开始学的特征值和特征向量的时候只是知道了定义和式子&#xff0c;并没有理解其内在的含义和应用&#xff0c;这段时间整理了相关的内容&#xff0c;跟大家…
阅读更多...
特征值和特征向量的几何含义理解

特征值和特征向量的几何含义理解

在刚开始学的特征值和特征向量的时候只是知道了定义和式子&#xff0c;并没有理解其内在的含义和应用&#xff0c;这段时间整理了相关的内容&#xff0c;跟大家分享一下&#xff1b; 首先我们先把特征值和特征向量的定义复习一下&#xff1a; 定义&#xff1a; 设A是n阶矩阵&am…
阅读更多...
线性代数之——特征值和特征向量

线性代数之——特征值和特征向量

线性方程 A x b Axb Axb 是稳定状态的问题&#xff0c;特征值在动态问题中有着巨大的重要性。 d u / d t A u du/dtAu du/dtAu 的解随着时间增长、衰减或者震荡&#xff0c;是不能通过消元来求解的。接下来&#xff0c;我们进入线性代数一个新的部分&#xff0c;基于 A x …
阅读更多...
特征值和特征向量概述-面试必问3(含特征值、向量意义)

特征值和特征向量概述-面试必问3(含特征值、向量意义)

特征值和特征向量&#xff08;Eigenvalues and eigenvectors&#xff09; 在线性代数中&#xff0c;一个线性变换的特征向量&#xff08;eigenvector 或者 characteristic vector&#xff09;是一个非零向量。将线性变换应用在它上面&#xff0c;它最多以一个标量因子进行伸缩…
阅读更多...
java 如何实现深拷贝

java 如何实现深拷贝

1、什么叫Java浅拷贝&#xff1f;  浅拷贝是按位拷贝对象&#xff0c;它会创建一个新对象&#xff0c;这个对象有着原始对象属性值的一份精确拷贝。如果属性是基本类型&#xff0c;拷贝的就是基本类型的值&#xff1b;如果属性是内存地址&#xff08;引用类型&#xff09;&…
阅读更多...
单例模式之枚举实现

单例模式之枚举实现

如果你没有学过单例模式&#xff0c;请点击&#xff1a;确保对象的唯一性——单例模式。 有很多网友留言说我漏掉了一种非常重要的Java语言的单例模式实现方式——枚举。^_^ 这篇姗姗来迟的博文将弥补这个“巨大的”缺陷。^_^~~~~~~~~~~~ 在Java语言中&#xff0c;如果综合考虑…
阅读更多...
vue实现购物车功能

vue实现购物车功能

随着时代发展&#xff0c;网购成了人们必不可少的一部分&#xff0c;所以我们常常遇到要实现购物车功能&#xff0c;如下图&#xff0c;我们来分析一下 下图所示页面: 首先&#xff0c;我们通过ElementUI中的<el-table>标签来实现页面的呈现。 其次&#xff0c;我们可以看…
阅读更多...
css实现轮播图

css实现轮播图

轮播图&#xff1a;就是多张图片按照一定的时间和顺序依次从某个窗口来向用户展示图片 轮播图的实现代码&#xff1a; 1&#xff09;创建一个容器来进行轮播图的展示 这里的容器就是最外部的盒子 注意最外部盒子设置宽高时要与我们进行展示的图片的宽高保持一致&#xff0c…
阅读更多...
RabbitMQ实现延迟队列的方式

RabbitMQ实现延迟队列的方式

1.背景 最近在做类似拍卖系统的上架功能&#xff0c;卖家上架物品以后&#xff0c;例如到期时间24小时或者48小时&#xff0c;如果无竞拍者或者购买者&#xff0c;则物品自动下架到用户的邮件中。诸如电商用户下单&#xff0c;30分钟未支付&#xff0c;则自动取消订单&#xff…
阅读更多...
接口的实现详解

接口的实现详解

接口 接口就是定义的规则&#xff0c;规范。 声明类时需要使用的关键字时class&#xff0c;声明接口的关键字时interface&#xff1b; 接口本身就是抽象的&#xff0c;需要一个实现类去实现接口中定义的内容。 接口当中不能定义方法&#xff1a; 接口本身就是抽象的 所在我们…
阅读更多...
推荐文章

Copyright @ 2022~2023