MySQL 数据历史变更构建

2023年7月22日 · 阅读需 5 分钟

John Li

Chief Executive Officer

简述

CloudCanal 为满足用户对于核心数据风控需求，在 MySQL->Greenplum 链路支持了历史数据能力，构建的历史数据特点包括:

exactly once(有且仅有一次)
支持事务内多次变更，反复写入删除等特殊场景

本文通过一个实际案例简要介绍如何使用这个功能。

技术点

核心数据变更风控的必要性

用户数据、交易数据的变更追踪，常用方式是通过应用埋点、数据网关或代理、数据管理控制平台等方法进行，通过规则预防、事后止损两大方向控制风险。

通过数据库增量日志追踪数据变更历史，从风控角度属于事后止损方向，常见落地方案如准实时数据对账，数据历史算法分析与统计等。

CloudCanal 目前根据用户需求，支持 MySQL->Greenplum 链路数据实时变更历史追踪功能。

历史数据结构形态

历史数据构建过程中，幂等性能够大幅度便利用户数据分析，用户无需分辨相类似数据是否是业务层面变更还是系统原因造成的重复数据。为此，我们结合具体数据链路，构建了幂等历史数据表结构。表结构如下。

CREATE TABLE "public"."worker_stats" (
  "id" bigint NOT NULL ,
  "__action" varchar(64),
  "__seq_in_tx" bigint,
  "__gtid" varchar(255),
  ...other column...
  PRIMARY KEY ("id","__action","__seq_in_tx","__gtid")
) ;

其中 id 为原始表主键，可以为联合主键（也就会有多个字段），__action 为此变更操作类型，包含 INSERT/UPDATE/DELETE 3个候选值， __seq_in_tx 为本变更在事务中的序号，__gtid 表示事务号。以上几个字段，在历史数据表结构中构成联合主键，达到幂等目标。

操作示例

前置条件:

下载安装 CloudCanal 私有部署版本,使用参见快速上手文档
准备好 MySQL 数据库(本例版本为 8.0) 和 Greenplum 数据库(本例为阿里云 ADB for PG)
准备好 MySQL 示例表以及 Greenplum 中表,并造一些数据。

-- MySQL

CREATE TABLE `worker_stats` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `worker_id` bigint(20) NOT NULL,
  `cpu_stat` text,
  `mem_stat` text,
  `disk_stat` text,
  `col_new` varchar(255) NOT NULL DEFAULT '123',
  `unsigned_col` bigint(20) unsigned DEFAULT NULL,
  `new_col_col` varchar(123) DEFAULT NULL,
  `timestamp_col` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=503723 DEFAULT CHARSET=utf8mb4;

-- Greenplum

CREATE TABLE "public"."worker_stats" (
  "id" bigint NOT NULL ,
  "__action" varchar(64),
  "__seq_in_tx" bigint,
  "__gtid" varchar(255),
  "gmt_create" timestamp NOT NULL,
  "worker_id" bigint NOT NULL,
  "cpu_stat" text,
  "mem_stat" text,
  "disk_stat" text,
  "col_new" varchar(255) NOT NULL DEFAULT '123',
  "unsigned_col" bigint DEFAULT NULL,
  "new_col_col" varchar(123) DEFAULT NULL,
  "timestamp_col" timestamp NOT NULL,
  PRIMARY KEY ("id","__action","__seq_in_tx","__gtid")
) ;