关于数据库:实现MySQL同步数据到ES构建宽表

5次阅读

共计 5543 个字符，预计需要花费 14 分钟才能阅读完成。

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 索引，实现附加信息全量和增量的数据同步。

# 创立产品信息表
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='产品标签关系表';

# 填充产品信息
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);

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

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

初始化的我的项目须要手工配置一下 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>

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;}
}

        @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;
        }

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

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

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

此时查问产品数据，失去后果

可去掉主动启动工作选项，以便于单步追踪调试

Tips: 只配置减少操作，不要配置编辑和删除，否则可能造成对数据的误删；编辑和删除操作，只最好应用 ES 调用的形式进行解决；减少操作最好不要应用 ES 调用的形式解决，会引起高并发问题。

Tips: 创立工作时如果不上传自定义代码包，之后将无奈上传，除非重建工作。上传自定义代码，意味着创立非凡类型的工作，而后才会呈现非凡的选项进行字段映射。

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

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

Tips: product_id 字段必须做映射，否则即便配置了 _id 信息，仍旧无奈失常执行，会疏忽 product_id 字段的值。

自定义代码在开发阶段最麻烦的事件是如何高效进行调试，CloudCanal 可能比拟敌对的让开发在本地间接调试代码逻辑。

工作详情 -> 参数批改

Tips：每次批改完参数信息之后，必须点击失效配置和重启工作；在工作详情配置中，也能够上传新的代码包，激活和重启工作后能够应用。

Tips: 设置好断点当前，须要先启动 CloudCanal 工作，再点击 debug 按钮，能力 Attach 到近程的 8787 端口；CloudCanal 会始终 pending，直到有 Attachment，才会继续执行，所以不须要单步跟踪调试时，肯定记得敞开调试模式，否则工作无奈执行。

CloudCanal 自定义代码可能拓展的能力具备不错的设想空间，咱们甚至能退出一些在线业务逻辑的解决，让业务需要可能更好的满足，同时配合社区版调试也很不便。心愿将来这块能力在便当性能，性能等层面有更好的体现。

CloudCanal 会一直提供一些预览的能力，包含新数据链路, 优化能力, 性能插件。本文所形容的自定义代码能力目前也处于内测阶段。如需体验，可增加咱们小助手 (微信号:suhuayue001) 进行理解和试用。

退出 CloudCanal 粉丝群把握一手音讯和获取更多福利，请增加咱们小助手微信：suhuayue001
CloudCanal- 收费好用的企业级数据同步工具，欢送品鉴。
理解更多产品能够查看官方网站：http://www.clougence.com
CloudCanal 社区：https://www.askcug.com/

正文完

数据库

发表至：数据库

2021-11-15

0

关于数据库:ApiPost-快速生成在线接口文档

关于数据库:DTCC-2022云原生数据库-PieCloudDB-全新-eMPP-架构是如何炼成的

关于数据库:连载如何掌握openGauss数据库核心技术秘诀五拿捏数据库安全3

关于数据库:MongoDB-最佳实践qbit

关于linux:oneinstack申请免费的R3-域名证书

关于数据库:实现MySQL同步数据到ES构建宽表

作者介绍

前言

场景形容

同步策略

实现步骤

1. MySQL 表构造初始化

2. MySQL 填充测试数据

3. ElasticSearch 索引创立（也能够应用 CloudCanal 构造迁徙）

4. 编写自定义代码

4.1 批改 MAVEN 配置

4.2 实现 TAG 类

4.3 实现 PROCESSOR 解决逻辑

4.4 编译自定义代码包

5. 创立 CloudCanal 工作

5.1 同步 PRODUCT 根底数据

5.2 扩大 PRODUCT TAG 数据

5.2.1 配置数据源和指标

5.2.2 配置规格

5.2.3 配置索引映射

5.2.4 上传自定义代码

5.2.5 配置字段映射

6. 同步后果

调试自定义代码

批改工作参数

配置 IntelliJ IDEA Debug 模式

总结

参加内测

站内搜索