作者介绍
Ceven，德勤乐融(北京)科技有限公司
邮箱：likailin@deqinyuerong.com
前言
CloudCanal 近期提供了自定义代码构建宽表能力，咱们第一工夫参加了该个性内测，成果不错。开发流程详见官网文档 《CloudCanal自定义代码实时加工》
能力特点包含：
- 灵便，反对反查打宽表，特定逻辑数据荡涤，对账，告警等场景
- 调试不便，通过工作参数配置主动关上 debug 端口，对接 IDE 调试
- SDK 接口清晰，提供丰盛的上下文信息，不便数据逻辑开发
本文基于咱们业务中的理论需要(MySQL -> ElasticSearch 宽表构建)，梳理一下具体的开发调试流程，心愿对大家有所帮忙。
场景形容
MySQL 善于关系型数据操作，咱们在其中存储了 product, tag, product_tag_mapping 表数据，用以示意产品和标签之间多对多关系。精简的数据结构如下：
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 解决逻辑
@Override public 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=true5. 创立 CloudCanal 工作
5.1 同步 PRODUCT 根底数据
全量增量同步 product 信息到 es_product 索引，在此就不做具体形容，详情请参考 CloudCanal 文档。
此时查问产品数据，失去后果
5.2 扩大 PRODUCT TAG 数据
5.2.1 配置数据源和指标
5.2.2 配置规格
可去掉主动启动工作选项，以便于单步追踪调试
5.2.3 配置索引映射
Tips: 只配置减少操作，不要配置编辑和删除，否则可能造成对数据的误删； 编辑和删除操作，只最好应用 ES 调用的形式进行解决； 减少操作最好不要应用 ES 调用的形式解决，会引起高并发问题。5.2.4 上传自定义代码
Tips: 创立工作时如果不上传自定义代码包，之后将无奈上传，除非重建工作。 上传自定义代码，意味着创立非凡类型的工作，而后才会呈现非凡的选项进行字段映射。5.2.5 配置字段映射
将 id 和 tag_id 调整为 “只订阅不同步”(老版本此处会显示为仅供自定义代码应用)，实现只订阅这两个字段，而不会真正写入到 ES 索引，而将 product_id 映射到对端的 id。
设置映射 _id，以指定指标 ES 索引中的 id 为 product_id
Tips: product_id 字段必须做映射，否则即便配置了 _id 信息，仍旧无奈失常执行，会疏忽 product_id 字段的值。6. 同步后果
调试自定义代码
自定义代码在开发阶段最麻烦的事件是如何高效进行调试，CloudCanal 可能比拟敌对的让开发在本地间接调试代码逻辑。
批改工作参数
工作详情->参数批改
Tips：每次批改完参数信息之后，必须点击失效配置和重启工作； 在工作详情配置中，也能够上传新的代码包，激活和重启工作后能够应用。配置 IntelliJ IDEA Debug 模式
Tips: 设置好断点当前，须要先启动 CloudCanal 工作，再点击 debug 按钮，能力 Attach 到近程的 8787 端口； CloudCanal 会始终 pending，直到有 Attachment，才会继续执行，所以不须要单步跟踪调试时，肯定记得敞开调试模式，否则工作无奈执行。总结
CloudCanal 自定义代码可能拓展的能力具备不错的设想空间，咱们甚至能退出一些在线业务逻辑的解决，让业务需要可能更好的满足，同时配合社区版调试也很不便。心愿将来这块能力在便当性能，性能等层面有更好的体现。
参加内测
CloudCanal 会一直提供一些预览的能力，包含新数据链路, 优化能力,性能插件。本文所形容的自定义代码能力目前也处于内测阶段。如需体验，可增加咱们小助手(微信号:suhuayue001)进行理解和试用。
退出CloudCanal粉丝群把握一手音讯和获取更多福利，请增加咱们小助手微信：suhuayue001
CloudCanal-收费好用的企业级数据同步工具，欢送品鉴。
理解更多产品能够查看官方网站： http://www.clougence.com
CloudCanal社区：https://www.askcug.com/