作者介紹
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=true
5. 創建 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)進行了解和試用。