数据同步一般分为两种方式：全量和增量。增量数据是一类典型的流数据，基于日志的增量同步几乎已经是所有数据库的标配，它可以减少常规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) 交流。