如何用gpss实现MySQL到Greenplum的增量同步

2022-06-08 00:00:00 数据 操作 消息 增量 复制

数据同步一般分为两种方式:全量和增量。增量数据是一类典型的流数据,基于日志的增量同步几乎已经是所有数据库的标配,它可以减少常规ETL工作对系统带来的影响,并大大降低数据的延迟。作为Greenplum的流计算引擎,Greenplum Stream Server(gpss)能将不同源端的增量数据同步到Greenplum中。为更好的支持这一应用场景,即将发布的gpss 1.3.6 对增量同步的功能做了增强。

Greenplum Stream Server(简称gpss),是Greenplum的下一代数据加载解决方案,相比于gpfdist,GPSS会提供流数据支持及API接口,有更好的扩展性,支持更丰富的功能,并开放更细粒度的任务控制接口。在即将发布gpss 1.3.6 中,对增量同步所做的的功能增强包括:

可以根据指定的递增排序字段,确保新的消息生效

Merge可支持insert,update和delete三种操作

本文将以MySQL为例,简要介绍下gpss如何实现向Greenplum的增量同步。

测试环境

  • MySQL 8.0.13
  • Maxwell 1.25.0
  • Kafka 2.2.2
  • Greenplum 6.4.0
  • GPSS 1.3.6

我们要完成的工作是:

  • 通过Maxwell监听MySQL中binlog的增量变化(略)

  • 将增量数据以json的格式发送到kafka中(略)

  • 利用gpss解析kafka中的json消息

  • 将变化的数据更新到Greenplum的目标表中

MySQSL和Maxwell的配置和使用,本文将不做深入介绍,大家可以自行访问文章链接阅读学习,访问相关文章请点击文章底部的“阅读原文”。

测试数据简介

测试使用的表在MySQL中定义如下:

create table t_update_delete_0 (k1 decimal,
               k2 text,
               v1 decimal,
               v2 decimal,
               v3 text,
               c1 decimal,
               c2 text);

复制代码

其中 k1 和 k2 列为键,用来标识一条记录, v1, v2, v3 为每次更新的数据。

在源端分别对这个表进行了insert,update和delete操作,每个语句为单独的transaction。

Insert语句为:


insert into t_update_delete_0 (k1,k2,v1,v2,v3,c1,c2) 
        values (1,'k_1', 1, 3, 'v_1', 1, 'c1');

复制代码

Update语句为:

update t_update_delete_0 set v1=100,v2=300,v3='v_100' where k1='1' and k2='k_1';

Delete语句为:

delete from t_update_delete_0 where k1='1' and k2='k_1';
复制代码

Kafka的消息格式

Maxwell可以将捕获到binlog解析为json格式并发送到kafka,不同的操作生成的Kafka消息有细微的区别。为了将这些消息正确的恢复到Greenplum中,我们先对这三种类型的消息进行简单的分析。

Insert时生成的消息示例如下:

{
  "database": "test",
  "table": "t_update_delete_0",
  "type": "insert",
  "ts": 1586956781,
  "xid": 1398209,
  "commit": true,
  "data": {
    "k1": 41,
    "k2": "k_41",
    "v1": 818,
    "v2": 2454,
    "v3": "v_818",
    "c1": 41,
    "c2": "c_41"
  }
}
复制代码

database和table表示源表的表名,ts和xid字段用于表示消息的顺序,type和data表示执行的操作及对应的数据。这些是所有消息类型通用的。

Delete生成的消息如下,type为"delete",同时data中包含了完整的内容。

{
  "database": "test",
  "table": "t_update_delete_0",
  "type": "delete",
  "ts": 1586956781,
  "xid": 1398195,
  "commit": true,
  "data": {
    "k1": 44,
    "k2": "k_44",
    "v1": 744,
    "v2": 2232,
    "v3": "v_744",
    "c1": 44,
    "c2": "c_44"
  }
}
复制代码

Update除了包含新数据外,还包含了更新之前的数据(old),这里我们只需要新数据就够了。

{
  "database": "test",
  "table": "t_update_delete_0",
  "type": "update",
  "ts": 1586956707,
  "xid": 1281915,
  "commit": true,
  "data": {
    "k1": 99,
    "k2": "k_99",
    "v1": 798,
    "v2": 2394,
    "v3": "v_798",
    "c1": 99,
    "c2": "c_99"
  },
  "old": {
    "v1": 800,
    "v2": 2400,
    "v3": "v_800"
  }
}
复制代码

根据生成的消息,我们需要执行如下操作:

根据ts和xid对数据进行排序

根据k1和k2进行匹配

对type为delete的列执行删除操作

对其它type类型执行Merge(upsert)操作

执行gpss的Kafka JOB

Greenplum中的定义包含了排序的字段,用来区分消息更新的先后顺序,定义如下:

create table t_update_delete_0 (k1 decimal,
               k2 text,
               v1 decimal,
               v2 decimal,
               v3 text,
               c1 decimal,
               c2 text,
               ts decimal,
               xid decimal,
               del_mark boolean);

复制代码

根据数据同步的需求,gpss需要的yaml配置文件如下:

DATABASE: test
USER: gpadmin
HOST: mdw
PORT: 5432
VERSION: 2
KAFKA:
   INPUT:
      SOURCE:
        BROKERS: kafkahost:9092
        TOPIC: test
      VALUE:
        COLUMNS:
          - NAME: c1
            TYPE: json
        FORMAT: json
      ERROR_LIMIT: 100
   OUTPUT:
      MODE: MERGE
      MATCH_COLUMNS:
        - k1
        - k2
      UPDATE_COLUMNS:
        - v1
        - v2
        - v3
      ORDER_COLUMNS:
        - ts
        - xid
      DELETE_CONDITION: del_mark
      TABLE: t_update_delete_0
      MAPPING:
         k1 : (c1->'data'->>'k1')::decimal
         k2 : (c1->'data'->>'k2')::text
         v1 : (c1->'data'->>'v1')::decimal
         v2 : (c1->'data'->>'v2')::decimal
         v3 : (c1->'data'->>'v3')::text
         c1 : (c1->'data'->>'c1')::decimal
         c2 : (c1->'data'->>'c2')::text
         ts : (c1->>'ts')::decimal
         xid: (c1->>'xid')::decimal
         del_mark: (c1->>'type')::text = 'delete'
   COMMIT:
      MINIMAL_INTERVAL: 2000
复制代码

几个主要的配置含义如下:

  • ORDER_COLUMNS:递增排序的字段,每个batch中,gpss会使用ORDER_COLUMNS大的消息内容对目标表进行操作。
  • DELETE_CONDITION:软删除标记,gpss会删除包含DELETE_CONDITION字段的匹配记录
  • MATCH_COLUMNS:记录的标识,也就是键(candidate key)
  • UPDATE_COLUMNS:需要更新的列

概括下来,gpss执行的步骤为:

  1. 在一个batch中,针对MATCH_COLUMNS相同的所有记录,先根据ORDER_COLUMNS去重

  2. 目标表中存在MATCH_COLUMNS匹配的记录时,根据UPDATE_CONDITION或者DELETE_CONDITION执行更新或者删除操作

  3. 目标表中不存在时匹配记录时,执行插入操作。

(由于有去重操作,为保证不丢失数据,在UPDATE时,Kafka的消息中需要包含整行的数据,而不仅仅是更新部分的数据。)

配置文件准备好后,我们通过gpkafka来执行加载:

gpkafka load mysql.yaml
复制代码

gpkafka便会从kafka中拉取对应的消息,按照设定的操作将Kafka中的增量数据同步到目标表中。

小结

本文简单介绍了如何用gpss从MySQL进行增量同步,其它数据库(例如Oracle,SQL Server等)也都可以利用类似的方案实现同步。不同的消息类型需要不同的处理逻辑,gpss的配置文件中有很多可以进行后处理的部分,更详细的内容可以参考官方文档:gpdb.docs.pivotal.io/streaming-s… offset排序等;欢迎大家使用,反馈,指导。 也欢迎大家前往askGP(ask.greenplum.cn) 交流。

相关文章