前言
CloudCanal 近期提供了自定義代碼構(gòu)建寬表能力，我們第一時間參與了該特性內(nèi)測，效果不錯。開發(fā)流程詳見官方文檔 《CloudCanal自定義代碼實時加工》https://www./topic/211
能力特點包括：
靈活，支持反查打?qū)挶?，特定邏輯?shù)據(jù)清洗，對賬，告警等場景
調(diào)試方便，通過任務(wù)參數(shù)配置自動打開 debug 端口，對接 IDE 調(diào)試
SDK 接口清晰，提供豐富的上下文信息，方便數(shù)據(jù)邏輯開發(fā)
本文基于我們業(yè)務(wù)中的實際需求(MySQL -> ElasticSearch 寬表構(gòu)建)，梳理一下具體的開發(fā)調(diào)試流程，希望對大家有所幫助。
場景描述
MySQL 擅長關(guān)系型數(shù)據(jù)操作，我們在其中存儲了 product, tag, product_tag_mapping 表數(shù)據(jù)，用以表示產(chǎn)品和標(biāo)簽之間多對多關(guān)系。精簡的數(shù)據(jù)結(jié)構(gòu)如下：
ElasticSearch 擅長搜索，但是并不支持不同索引間的聯(lián)合查詢, 所以構(gòu)造寬表是業(yè)界剛需。我們存儲其上的產(chǎn)品索引結(jié)構(gòu)如下：
PUT es_product
{
'mappings' : {
'properties' : {
'id' : {
'type' : 'integer'
},
'name' : {
'type' : 'text'
},
'tags' : {
'type' : 'nested',
'properties' : {
'id' : {
'type' : 'integer'
},
'name' : {
'type' : 'text'
}
}
}
}
}
}
同步策略
CloudCanal 在 同步 MySQL -> ElasticSearch 數(shù)據(jù)過程中，會兼顧全量和增量兩種情況，我們可以創(chuàng)建兩個獨立的任務(wù)，分別同步產(chǎn)品的基礎(chǔ)信息和附加信息（即標(biāo)簽信息）。
基礎(chǔ)信息任務(wù)
附加信息任務(wù)
實現(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 '名稱',
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 '名稱',
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 填充測試數(shù)據(jù)
# 填充產(chǎn)品信息
INSERT INTO `product` (`name`)
VALUES
('product_1');
# 填充標(biāo)簽信息
INSERT INTO `tag` (`name`)
VALUES
('tag_1'),
('tag_2');
# 填充產(chǎn)品標(biāo)簽關(guān)系信息
INSERT INTO `product_tag_mapping` (`product_id`, `tag_id`)
VALUES
(1, 1);
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. 編寫自定義代碼
自定義代碼的項目基于 maven 構(gòu)建，可以參考 示例項目 cloudcanal-sdk-demos（https:///clougence/cloudcanal-sdk-demos/tree/master）
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 實現(xiàn) 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 實現(xiàn) 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 的嵌套結(jié)構(gòu)會被認(rèn)為是獨立的文檔，故需要填充舊的數(shù)據(jù)
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)) {
// 增量創(chuàng)建的 product_tag_mapping 處于內(nèi)存中，無法通過 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 編譯自定義代碼包
執(zhí)行如下命令編譯生成自定義代碼包, 之后會在 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 索引，在此就不做具體描述，詳情請參考 CloudCanal 文檔。
此時查詢產(chǎn)品數(shù)據(jù)，得到結(jié)果
5.2 擴(kuò)展 PRODUCT TAG 數(shù)據(jù)
5.2.1 配置數(shù)據(jù)源和目標(biāo)
5.2.2 配置規(guī)格
可去掉自動啟動任務(wù)選項，以便于單步追蹤調(diào)試
5.2.3 配置索引映射
Tips: 只配置增加操作，不要配置編輯和刪除，否則可能造成對數(shù)據(jù)的誤刪；
編輯和刪除操作，只最好使用 ES 調(diào)用的方式進(jìn)行處理；
增加操作最好不要使用 ES 調(diào)用的方式處理，會引起高并發(fā)問題。
5.2.4 上傳自定義代碼
Tips: 創(chuàng)建任務(wù)時如果不上傳自定義代碼包，之后將無法上傳，除非重建任務(wù)。
上傳自定義代碼，意味著創(chuàng)建特殊類型的任務(wù)，然后才會出現(xiàn)特殊的選項進(jìn)行字段映射。
5.2.5 配置字段映射
將 id 和 tag_id 調(diào)整為 “只訂閱不同步”(老版本此處會顯示為僅供自定義代碼使用)，實現(xiàn)只訂閱這兩個字段，而不會真正寫入到 ES 索引，而將 product_id 映射到對端的 id。
設(shè)置映射 _id，以指定目標(biāo) ES 索引中的 id 為 product_id
Tips: product_id 字段必須做映射，否則即使配置了 _id 信息，依舊無法正常執(zhí)行，會忽略 product_id 字段的值。
6. 同步結(jié)果
調(diào)試自定義代碼
自定義代碼在開發(fā)階段最麻煩的事情是如何高效進(jìn)行調(diào)試，CloudCanal 能夠比較友好的讓開發(fā)在本地直接調(diào)試代碼邏輯。
修改任務(wù)參數(shù)
任務(wù)詳情->參數(shù)修改
Tips：每次修改完參數(shù)信息之后，必須點擊生效配置和重啟任務(wù)；
在任務(wù)詳情配置中，也可以上傳新的代碼包，激活和重啟任務(wù)后可以使用。
配置 IntelliJ IDEA Debug 模式
Tips: 設(shè)置好斷點以后，需要先啟動 CloudCanal 任務(wù)，再點擊 debug 按鈕，才能 Attach 到遠(yuǎn)程的 8787 端口；
CloudCanal 會一直 pending，直到有 Attachment，才會繼續(xù)執(zhí)行，所以不需要單步跟蹤調(diào)試時，一定記得關(guān)閉調(diào)試模式，否則任務(wù)無法執(zhí)行。
總結(jié)
CloudCanal 自定義代碼能夠拓展的能力具有不錯的想象空間，我們甚至能加入一些在線業(yè)務(wù)邏輯的處理，讓業(yè)務(wù)需求能夠更好的滿足，同時配合社區(qū)版調(diào)試也很方便。希望未來這塊能力在便利功能，性能等層面有更好的表現(xiàn)。
參與內(nèi)測
