之前我们在《如何用gpss实现MySQL到Greenplum的增量同步》中详细介绍了MySQL到Greenplum增量同步的实现步骤。今天将给大家讲一讲Oracle到Greenplum又是如何实现的。
Oracle数据库虽然在OLTP领域仍有着毋庸置疑的优势地位，但在OLAP领域与Greenplum则是差距显著。如今已经有越来越多的分析型业务从Oracle迁移到Greenplum，在《如何从Oracle迁移到Greenplum》系列文章中，详细介绍了业务的迁移的最佳实践；而数据迁移中最核心的就是如何实现数据的实时增量同步。
对增量同步而言，gpss作为一个流计算框架，与源端是解耦的，因此只要Kafka topic中的消息，包含足够的信息，gpss都可以提取变化的数据并重放到gp中。之前介绍了如何利用gpss同步来自Maxwell和Mysql的增量数据，这里再以Oracle Golden Gate为例，介绍如何实时同步来自Oracle的增量数据。
1 测试环境
- Oracle
- Oracle Golden Gate
- Kafka 2.2.2
- Greenplum 6.4.0
- GPSS 1.3.6
我们要完成的工作是:
- 通过GoldenGate将Oracle中的增量数据以json格式发送到Kafka （略）
- 利用gpss解析kafka中的json消息
- 将变化的数据更新到Greenplum的目标表中
2 测试数据简介
测试使用的表在Oracle中定义如下：
CREATE TABLE SIEBEL_TEST.TEST_POC( ID numeric, NAME varchar2 (50), BIRTHDAY date)其中 ID 列为键，用来唯一标识一条记录， NAME 和 BIRTHDAY 为更新字段。
在源端分别对这个表进行了insert，update和delete操作。
Insert语句为:
insert into test_poc values (1, 'Igor', '01-JAN-2000');Update语句为:
update test_poc set birthday=add_months(birthday,1) where id <3;Delete语句为:
delete from test_poc where id=3;3 Kafka的消息格式
接下来我们对Golden Gate的这三种类型的消息进行简单的分析。
Insert时生成的消息示例如下：
{ "table": "SIEBEL_TEST.TEST_POC", "op_type": "I", "op_ts": "2019-11-21 10:05:34.000000", "current_ts": "2019-11-21T11:05:37.823000", "pos": "00000000250000058833", "tokens": { "TK_OPTYPE": "INSERT", "SCN": "" }, "after": { "ID": 1, "NAME": "Igor", "BIRTHDAY": "2000-01-01 00:00:00" }}Table表示源表的表名，current_ts表示操作发生的时间，这里我们用它做排序；op_type和after表示执行的操作及对应的数据。
Delete生成的消息如下，op_type为"D"，同时before中包含了完整的内容。
{ "table": "SIEBEL_TEST.TEST_POC", "op_type": "D", "op_ts": "2019-11-21 10:13:19.000000", "current_ts": "2019-11-21T11:13:23.060002", "pos": "00000000250000059999", "tokens": { "TK_OPTYPE": "DELETE", "SCN": "" }, "before": { "ID": 3, "NAME": "Gianluca", "BIRTHDAY": "2002-01-01 00:00:00" }}Update除了包含新数据(after)外，还包含了更新之前的数据(before), op_type类型为'U'。
{ "table": "SIEBEL_TEST.TEST_POC", "op_type": "U", "op_ts": "2019-11-21 10:13:19.000000", "current_ts": "2019-11-21T11:13:23.060000", "pos": "00000000250000059561", "tokens": { "TK_OPTYPE": "SQL COMPUPDATE", "SCN": "" }, "before": { "ID": 1, "NAME": "Igor", "BIRTHDAY": "2000-01-01 00:00:00" }, "after": { "ID": 1, "NAME": "Igor", "BIRTHDAY": "2000-02-01 00:00:00" }}根据生成的消息，我们需要执行如下操作：
- 根据id对消息去重
- 根据ts对消息排序
- 对op_type为D的列执行删除操作
- 对其它type类型执行Merge(upsert)操作
4 执行gpss的Kafka JOB
Greenplum中的定义包含了用于排序的字段ts，用来区分消息更新的先后顺序，定义如下：
CREATE TABLE test_poc( id numeric, name varchar (50), birthday date, ts timestamp);根据数据同步的需求，gpss需要的yaml配置文件如下：
DATABASE: testUSER: gpadminHOST: mdwPORT: 5432VERSION: 2KAFKA: INPUT: SOURCE: BROKERS: kafkahost:9092 TOPIC: oggpoc VALUE: COLUMNS: - NAME: c1 TYPE: json FORMAT: json ERROR_LIMIT: 100 OUTPUT: MODE: MERGE MATCH_COLUMNS: - id UPDATE_COLUMNS: - name - birthday ORDER_COLUMNS: - ts DELETE_CONDITION: c1->>'op_type' = 'D' TABLE: test_poc MAPPING: - NAME: id EXPRESSION: | CASE WHEN ((c1->'after')::json is not null) THEN (c1->'after'->>'ID')::integer ELSE (c1->'before'->>'ID')::integer end - NAME: name EXPRESSION: | CASE WHEN ((c1->'after')::json is not null) THEN (c1->'after'->>'NAME')::text ELSE null end - NAME: birthday EXPRESSION: | CASE WHEN ((c1->'after')::json is not null) THEN (c1->'after'->>'BIRTHDAY')::date ELSE null end - NAME: ts EXPRESSION: (c1->>'current_ts')::timestamp COMMIT: MINIMAL_INTERVAL: 2000相关字段的含义和gpss实际执行的操作可参见参考文献[2]，这里着重介绍下有区别的的地方，也就是由于insert和update操作的实际内容包含在after中，而delete的内容包含在before中，每个字段的内容需要额外的判断逻辑：有after时读取after中的内容，否则读取before中的内容。
此外需要注意的是，当新消息的ORDER_COLUMNS的内容有重复时，gpss会把所有的包含重复内容的行，都记录到目标表中；这样的主要目的是为了避免数据丢失。因此在实际使用时一定要确保排序地段的唯一性。
配置文件准备好后，我们通过gpkafka来执行加载:
gpkafka load ogg.yamlgpkafka便会从kafka中拉取对应的消息，按照设定的操作将Kafka中的增量数据同步到目标表中。
5 小结
这里简单介绍了如何用gpss从Kafka消费GoldenGate生成的Oracle增量数据进行同步，其它数据库及CDC工具（例如Informatica，StreamSet，NIFI等）也都可以利用类似的方案实现同步。今后会做更多的相关介绍，欢迎大家试用，反馈，指导。
6 参考文献
- https://github.com/pdeemea/ka...
- 如何用gpss实现MySQL到Greenplum的增量同步
- https://gpdb.docs.pivotal.io/...