前言

CloudCanal 近期提供了自定義代碼構(gòu)建寬表能力，我們第一時(shí)間參與了該特性?xún)?nèi)測(cè)，效果不錯(cuò)。開(kāi)發(fā)流程詳見(jiàn)官方文檔 《CloudCanal自定義代碼實(shí)時(shí)加工》https://www./topic/211

能力特點(diǎn)包括：

• 靈活，支持反查打?qū)挶恚囟ㄟ壿嫈?shù)據(jù)清洗，對(duì)賬，告警等場(chǎng)景

• 調(diào)試方便，通過(guò)任務(wù)參數(shù)配置自動(dòng)打開(kāi) debug 端口，對(duì)接 IDE 調(diào)試

• SDK 接口清晰，提供豐富的上下文信息，方便數(shù)據(jù)邏輯開(kāi)發(fā)

本文基于我們業(yè)務(wù)中的實(shí)際需求(MySQL -> ElasticSearch 寬表構(gòu)建)，梳理一下具體的開(kāi)發(fā)調(diào)試流程，希望對(duì)大家有所幫助。

場(chǎng)景描述

MySQL 擅長(zhǎng)關(guān)系型數(shù)據(jù)操作，我們?cè)谄渲写鎯?chǔ)了 product, tag, product_tag_mapping 表數(shù)據(jù)，用以表示產(chǎn)品和標(biāo)簽之間多對(duì)多關(guān)系。精簡(jiǎn)的數(shù)據(jù)結(jié)構(gòu)如下：

ElasticSearch 擅長(zhǎng)搜索，但是并不支持不同索引間的聯(lián)合查詢(xún), 所以構(gòu)造寬表是業(yè)界剛需。我們存儲(chǔ)其上的產(chǎn)品索引結(jié)構(gòu)如下：

同步策略

CloudCanal 在 同步 MySQL -> ElasticSearch 數(shù)據(jù)過(guò)程中，會(huì)兼顧全量和增量兩種情況，我們可以創(chuàng)建兩個(gè)獨(dú)立的任務(wù)，分別同步產(chǎn)品的基礎(chǔ)信息和附加信息（即標(biāo)簽信息）。

• 基礎(chǔ)信息任務(wù)

• 使用基本的映射關(guān)系，將 MySQL 中的 product 數(shù)據(jù)表，映射到 es_product 索引中，即可保證全量和增量的數(shù)據(jù)同步。

• 附加信息任務(wù)

• 創(chuàng)建 CloudCanal 任務(wù)將 MySQL 中的 product_tag_mapping 數(shù)據(jù)表映射到 es_product 索引中，同步過(guò)程中反查源數(shù)據(jù)庫(kù)中的 tag 信息，構(gòu)造寬表數(shù)據(jù)，填充進(jìn) es_product 索引，實(shí)現(xiàn)附加信息全量和增量的數(shù)據(jù)同步。

實(shí)現(xiàn)步驟

1. MySQL 表結(jié)構(gòu)初始化

# 創(chuàng)建產(chǎn)品信息表
CREATE TABLE `product` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(64) COLLATE utf8_unicode_ci NOT NULL DEFAULT '' COMMENT '名稱(chēng)',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='產(chǎn)品信息記錄表';

# 創(chuàng)建標(biāo)簽信息表
CREATE TABLE `tag` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(64) COLLATE utf8_unicode_ci NOT NULL DEFAULT '' COMMENT '名稱(chēng)',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='標(biāo)簽信息記錄表';

# 創(chuàng)建產(chǎn)品標(biāo)簽關(guān)系表
CREATE TABLE `product_tag_mapping` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `product_id` bigint(20) unsigned NOT NULL DEFAULT '0' COMMENT '產(chǎn)品ID',
  `tag_id` bigint(20) unsigned NOT NULL DEFAULT '0' COMMENT '標(biāo)簽ID',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='產(chǎn)品標(biāo)簽關(guān)系表';

2. MySQL 填充測(cè)試數(shù)據(jù)

3. ElasticSearch 索引創(chuàng)建（也可以使用 CloudCanal 結(jié)構(gòu)遷移）

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

4. 編寫(xiě)自定義代碼

自定義代碼的項(xiàng)目基于 maven 構(gòu)建，可以參考 示例項(xiàng)目 cloudcanal-sdk-demos（https:///clougence/cloudcanal-sdk-demos/tree/master）

4.1 修改 MAVEN 配置

初始化的項(xiàng)目需要手工配置一下 pom.xml 文件，將 sdk 指向本地目錄文件，代碼片段如下

4.2 實(shí)現(xiàn) TAG 類(lèi)

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 實(shí)現(xiàn) PROCESSOR 處理邏輯

4.4 編譯自定義代碼包

執(zhí)行如下命令編譯生成自定義代碼包, 之后會(huì)在 target 目錄中生成 jar 文件

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

5. 創(chuàng)建 CloudCanal 任務(wù)

5.1 同步 PRODUCT 基礎(chǔ)數(shù)據(jù)

全量增量同步 product 信息到 es_product 索引，在此就不做具體描述，詳情請(qǐng)參考 CloudCanal 文檔。

此時(shí)查詢(xún)產(chǎn)品數(shù)據(jù)，得到結(jié)果

5.2 擴(kuò)展 PRODUCT TAG 數(shù)據(jù)

5.2.1 配置數(shù)據(jù)源和目標(biāo)

5.2.2 配置規(guī)格

可去掉自動(dòng)啟動(dòng)任務(wù)選項(xiàng)，以便于單步追蹤調(diào)試

5.2.3 配置索引映射

5.2.4 上傳自定義代碼

Tips: 創(chuàng)建任務(wù)時(shí)如果不上傳自定義代碼包，之后將無(wú)法上傳，除非重建任務(wù)。
      上傳自定義代碼，意味著創(chuàng)建特殊類(lèi)型的任務(wù)，然后才會(huì)出現(xiàn)特殊的選項(xiàng)進(jìn)行字段映射。

5.2.5 配置字段映射

將 id 和 tag_id 調(diào)整為 “只訂閱不同步”(老版本此處會(huì)顯示為僅供自定義代碼使用)，實(shí)現(xiàn)只訂閱這兩個(gè)字段，而不會(huì)真正寫(xiě)入到 ES 索引，而將 product_id 映射到對(duì)端的 id。

設(shè)置映射 _id，以指定目標(biāo) ES 索引中的 id 為 product_id

6. 同步結(jié)果

調(diào)試自定義代碼

自定義代碼在開(kāi)發(fā)階段最麻煩的事情是如何高效進(jìn)行調(diào)試，CloudCanal 能夠比較友好的讓開(kāi)發(fā)在本地直接調(diào)試代碼邏輯。

修改任務(wù)參數(shù)

任務(wù)詳情->參數(shù)修改

Tips：每次修改完參數(shù)信息之后，必須點(diǎn)擊生效配置和重啟任務(wù)；
      在任務(wù)詳情配置中，也可以上傳新的代碼包，激活和重啟任務(wù)后可以使用。

配置 IntelliJ IDEA Debug 模式

總結(jié)

CloudCanal 自定義代碼能夠拓展的能力具有不錯(cuò)的想象空間，我們甚至能加入一些在線(xiàn)業(yè)務(wù)邏輯的處理，讓業(yè)務(wù)需求能夠更好的滿(mǎn)足，同時(shí)配合社區(qū)版調(diào)試也很方便。希望未來(lái)這塊能力在便利功能，性能等層面有更好的表現(xiàn)。

參與內(nèi)測(cè)