DataHub：OGG插件介绍

OGG采集工具

一、背景介绍

OGG（Oracle GoldenGate）是一个基于日志的结构化数据备份工具，一般用于Oracle数据库之间的主从备份以及Oracle数据库到其他数据库（DB2, MySQL等）的同步。下面是Oracle官方提供的一个OGG的整体架构图，从图中可以看出OGG的部署分为源端和目标端两部分组成，主要有Manager，Extract，Pump，Collector，Replicat这么一些组件。

• Manager：在源端和目标端都会有且只有一个Manager进程存在，负责管理其他进程的启停和监控等；

• Extract：负责从源端数据库表或者事务日志中捕获数据，有初始加载和增量同步两种模式可以配置，初始加载模式是直接将源表数据同步到目标端，而增量同步就是分析源端数据库的日志，将变动的记录传到目标端，本文介绍的是增量同步的模式；

• Pump：Extract从源端抽取的数据会先写到本地磁盘的Trail文件，Pump进程会负责将Trail文件的数据投递到目标端；

• Collector：目标端负责接收来自源端的数据，生成Trail文件

• Replicat：负责读取目标端的Trail文件，转化为相应的DDL和DML语句作用到目标数据库，实现数据同步。

本文介绍的Oracle数据同步是通过OGG的DataHub插件实现的,该DataHub插件在架构图中处于Replicat的位置，会分析Trail文件，将数据的变化记录写入DataHub中，可以使用流计算对datahub中的数据进行实时分析，也可以将数据归档到MaxCompute中进行离线处理。当前插件版本为2.0.5，用户可以直接在文末下载最新版，也可以在github上下载源码自行打包。

二、支持数据库版本

三、使用场景

这里以Oracle数据为例，展示一个简单的示例来说明数据的使用方法，例如我们在数据库有一张商品订单表orders（oid int, pid int, num int），该表有三列，分别为订单ID, 商品ID和商品数量。将这个表通过OGG Datahub进行增量数据同步之前，我们需要先将源表已有的数据通过DataX同步到MaxCompute中。增量同步的关键步骤如下：（1）在DataHub上创建相应的Topic，Topic的schema为（string record_id, string optype, string readtime, bigint oid_before, bigint oid_after, bigint pid_before, bigint pid_after, bigint num_before, bigint num_after）；（2）OGG Datahub的插件按照上述的安装流程部署配置，其中列的Mapping配置如下：

<ctypeColumn>optype</ctypeColumn>
<ctimeColumn>readtime</ctimeColumn>
<cidColumn>record_id</cidColumn>
<columnMapping>
     <column src="oid" dest="oid_after" destOld="oid_before" isShardColumn="true"/>
     <column src="pid" dest="pid_after" destOld="pid_before"/>
     <column src="num" dest="num_after" destOld="num_before"/>
</columnMapping>

其中optype和readtime字段是记录数据库的数据变更类型和时间，optype有”I”, “D”, “U”三种取值，分别对应为“增”，“删”，“改”三种数据变更操作。（3）OGG Datahub插件部署好成功运行后，插件会源源不断的将源表的数据变更记录输送至datahub中，例如我们在源订单表中新增一条记录（1，2，1），datahub里收到的记录如下：

+--------+------------+------------+------------+------------+------------+------------+------------+------------+
| record_id | optype     | readtime   | oid_before | oid_after  | pid_before | pid_after  | num_before | num_after  |
+-------+------------+------------+------------+------------+------------+------------+------------+------------+
| 14810373343020000 |     I          | 2016-12-06 15:15:28.000141 | NULL       | 1          | NULL       | 2          | NULL       | 1   |

修改这条数据，比如把num改为20，datahub则会收到的一条变更数据记录，如下：

+-------+------------+------------+------------+------------+------------+------------+------------+------------+
| record_id | optype     | readtime   | oid_before | oid_after  | pid_before | pid_after  | num_before | num_after  |
+--------+------------+------------+------------+------------+------------+------------+------------+------------+
| 14810373343080000 |     U          | 2016-12-06 15:15:58.000253 | 1          | 1          | 2          | 2          | 1          | 20         |

(4)在源端OGG运行ggsci，然后运行stats dhext可以查看每个表的操作统计，在目标端同样可以使用，可以查看采集端和目标端操作统计是否一致。

实时计算

在前一天的离线计算的基础数据上，我们可以写一个StreamCompute流计算的分析程序，很容易地对数据进行实时汇总，例如实时统计当前总的订单数，每种商品的销售量等。处理思路就是对于每一条到来的变更数据，可以拿到变化的数值，实时更新统计变量即可。

离线处理

为了便于后续的离线分析，我们也可以将DataHub里的数据归档到MaxCompute中，在MaxCompute中创建相应Schema的表：

create table orders_log(record_id string, optype string, readtime string, oid_before bigint, oid_after bigint, pid_before bigint, pid_after bigint, num_before bigint, num_after bigint);

在DataHub上创建MaxCompute的数据归档，上述流入DataHub里的数据将自动同步到MaxCompute当中。建议将同步到MaxCompute中的数据按照时间段进行划分，比如每一天的增量数据都对应一个独立分区。这样当天的数据同步完成后，我们可以处理对应的分区，拿到当天所有的数据变更，而与和前一天的全量数据进行合并后，即可得到当天的全量数据。为了简单起见，先不考虑分区表的情况，以2016-12-06这天的增量数据为例，假设前一天的全量数据在表orders_base里面，datahub同步过来的增量数据在orders_log表中，将orders_base与orders_log做合并操作，可以得到2016-12-06这天的最终全量数据写入表orders_result中。这个过程可以在MaxCompute上用如下这样一条SQL完成。

INSERT OVERWRITE TABLE orders_result
SELECT t.oid, t.pid, t.num
FROM
(
     SELECT oid, pid, num, '0' x_record_id, 1 AS x_optype
     FROM
     orders_base 
     UNION ALL
     SELECT decode(optype,'D',oid_before,oid_after) AS oid
              , decode(optype,'D', pid_before,pid_after) AS pid
              , num_after AS num
              , record_id x_record_id
              , decode(optype, 'D', 0, 1) AS x_optype
     FROM
     orders_log
 ) t
JOIN
 (
     SELECT
     oid
     , pid
     , max(record_id) x_max_modified
     FROM
     (
     SELECT
     oid
     , pid
     , '0' record_id
     FROM
     orders_base UNION ALL SELECT
                      decode(optype,'D',oid_before,oid_after) AS oid
                      , decode(optype,'D', pid_before,pid_after) AS pid
                      , record_id
                      FROM
                      orders_log ) g
     GROUP BY oid , pid
 ) s
ON
t.oid = s.oid AND t.pid = s.pid AND t.x_record_id = s.x_max_modified AND t.x_optype <> 0;

四、参数详细介绍

configure

defaultOracleConfigure

defaultDatahubConfigure

constColumnMap 获取时间参数说明

例如：同步到MaxCompute时，想以天作为分区值，就可以添加配置<constColumnMap>ds=%Y%m%d</constColumnMap>，就会在DataHub中ds列中写入数据20200509。

mapping

columnMapping中的column

五、常见问题

问题排查常用文件:

datahub-ogg-plugin.log: 插件日志，DataHub插件进程打印的日志，一半以上的问题都可以通过查看这个日志报错来解决，在插件目录的log目录下。

ggserr.log: 监控日志。一般记录用户的操作记录，和是否启动成功，ogg/adapter主目录下。一般extract进程配置错误，或者缺少动态库文件报错记录都会在这个文件中。

dirrpt: report文件，可以查看该进程运行时的报错信息等（view report就是查看该文件），如果数据正常传输到目标端，ggserr.log和插件日志都没有报错，那么很可能错误记录在这个文件中。

dirchk: OGG的点位文件，记录了该进程的点位信息

datahub_ogg_plugin.chk: datahub插件自己的点位记录。

Q： 数据无法写入DataHub

A : 检查整条链路。

• 检查源端的dirdat是否有数据，如果没有那么是源端extract的问题

• 检查目标端dirdat是否有数据，如果没有那么是源端pump的问题

• 检查插件日志是否有记录，有可能是插件自己的点位导致直接忽略了数据,或者是插件有报错（最常见）,

• 如果插件日志完全没有数据，数据也被投递过来了，那么查看dirrpt下的日志，可能是目标端OGG自己的点位检查直接忽略掉数据了。

• 检查oracleSchema配置是否正确，有可能是oracleSchema和源端不匹配导致数据被忽略。

Q： 无法获取rowid，其余数据正常

A : 多是配置错误引起，需仔细检查配置。

• 检查版本，2.0.3才可以获取rowid。

• 检查源端extract是否正确配置token，TABLE OGG_TEST.*,tokens (TKN-ROWID=@GETENV('RECORD','rowid'));，之前就有用户将TKN-ROWID错写成TKN_ROWID，导致一致无法获取到rowid.

• 检查目标端configure.xml是否配置rowid， <rowIdColumn></rowIdColumn>

• 检查抽取数据(logdump)是否成功获取rowid，如下图

Q： Error in token clause for TKN-ROWID

报错信息：

2020-01-17 16:17:17  INFO    OGG-06509  Oracle GoldenGate Capture for Oracle, dhext.prm:  Using the following key columns for source table OGG_TEST.ORDERS: OID.
2020-01-17 16:17:17  INFO    OGG-01298  Oracle GoldenGate Capture for Oracle, dhext.prm:  Column function diagnostic message: could not find column "RECORD".
2020-01-17 16:17:17  ERROR   OGG-00369  Oracle GoldenGate Capture for Oracle, dhext.prm:  Error in token clause for TKN-ROWID.
2020-01-17 16:17:17  ERROR   OGG-01668  Oracle GoldenGate Capture for Oracle, dhext.prm:  PROCESS ABENDING.

A: 源端配置TABLE OGG_TEST.*,tokens (TKN-ROWID=@GETENV('RECORD','rowid'));之后，无法启动。

• 检查版本，2.0.3及以上版本才可以获取rowid

• 仔细检查配置是否与官方文档一致

• 如果一致可以尝试切换一下@GETENV('RECORD','rowid'))的单双引号

• 重启mgr

Q： dsOperation.getPosition():xxxxxxx old sendPosition is xxxxx, Skip this Operation, it maybe duplicated.

A: 多数情况是因为源端重启导致，表示Datahub插件点位重复了，但是这个经常会出现误判，而且有ogg自身的点位就足够了，所以可以尝试禁用掉插件点位（disableCheckPointFile设为true），不会对程序有影响，并且在接下来的版本中，我们也会完全弃用Datahub插件的点位。

Q： dirrpt下的日志报错 Unable to create operation: com.goldengate.atg.datasource.ConcurrentTransactionException: Multiple “Begin-Transaction” operations received for the same source transaction. Current transaction begin position: xxx(xxx); current trail position: xxx。

A: 源端事务冲突导致，需要排查源端Oracle是否有问题。

Q： 数据上传到DataHub之后有8小时的时间延迟

A: 修改javaue.properties中的参数goldengate.userexit.timestamp为utc+8，然后重启dhwriter。

Q： datahub插件解析后某些数据字段为空

A: 插件日志DEBUG模式下会打印全量数据，检查是否与源端数据匹配。

• 如果日志中打印数据正确，那可能是用户配置columnMapping错误导致数据缺失

• 如果日志中打印数据已经不匹配，那么检查OGG抽取数据是否正确，查看方法参考 logdump官方文档。rba信息可以在日志中看到。

Q：源端使用OGG12.3，插件解析trail文件失败

A: 首先建议使用OGG12.2,或者也可以目标端修改extract配置RMTTRAIL ./dirdat/st为RMTTRAIL ./dirdat/st format RELEASE 12.2。

Q： java.lang.NumberFormatException: for input string : “xxxx”

A: 检查所有的配置字段是否对应，对应关系可以参考Oracle类型与DataHub类型对应说明。之前有用户，把Oracle的字符串的destOld对应到DataHub的BIGINT字段，导致BIGINT解析失败。

Q： String index out of range: 10

报错信息：

2020-01-14 10:03:24 [main] ERROR OperationHandler - dirty data : INSERT on WF_YN_07.CJ_GLQX202001 [B] [columns: 27] [pos: 00000007260404799638 (seqno=726, rba=404799638)] [time: 2020-01-14 08:49:43.883741]
java.lang.StringIndexOutOfBoundsException: String index out of range: 10
    at java.lang.AbstractStringBuilder.setCharAt(AbstractStringBuilder.java:407)
    at java.lang.StringBuilder.setCharAt(StringBuilder.java:76)
    at com.aliyun.odps.ogg.handler.datahub.RecordBuilder.convertStrToMicroseconds(RecordBuilder.java:272)
    at com.aliyun.odps.ogg.handler.datahub.RecordBuilder.setTupleData(RecordBuilder.java:242)
    at com.aliyun.odps.ogg.handler.datahub.RecordBuilder.buildTupleRecord(RecordBuilder.java:172)
    at com.aliyun.odps.ogg.handler.datahub.RecordBuilder.buildRecord(RecordBuilder.java:82)
    at com.aliyun.odps.ogg.handler.datahub.operations.OperationHandler.processOperation(OperationHandler.java:50)
    at com.aliyun.odps.ogg.handler.datahub.operations.InsertOperationHandler.process(InsertOperationHandler.java:31)
    at com.aliyun.odps.ogg.handler.datahub.DatahubHandler.operationAdded(DatahubHandler.java:108)
    at oracle.goldengate.datasource.DsEventManager$3.send(DsEventManager.java:439)
    at oracle.goldengate.datasource.DsEventManager$EventDispatcher.distributeEvent(DsEventManager.java:231)
    at oracle.goldengate.datasource.DsEventManager.fireOperationAdded(DsEventManager.java:447)
    at oracle.goldengate.datasource.AbstractDataSource.fireOperationAdded(AbstractDataSource.java:464)
    at oracle.goldengate.datasource.UserExitDataSource.addOperationToTransactionAndFireEvent(UserExitDataSource.java:1337)
    at oracle.goldengate.datasource.UserExitDataSource.createOperation(UserExitDataSource.java:1305)
    at oracle.goldengate.datasource.UserExitDataSource.createOperation(UserExitDataSource.java:1046)

A: 用户把时间戳类型的数据映射到TIMESTAMP上，因为TIMESTAMP字段对应的Oracle字段类型一般默认为date或者timestamp，所以直接是按时间字符串”2020-01-10:10:00:00”来处理的，这个实际传过来是时间戳字符串，所以就会发生越界的情况。 解决方法：配置columnMapping中的isDateFormat="false"。

Q： 无法获取update的before值

A: 检查源端是否有配置GETUPDATEBEFORES。

Q： Oracle中有主键的表，OGG获取update操作，after的值为””

A: Oracle的表有没有主键，OGG获取的update操作的值是不同的：

• 对于没有主键表的update操作，如果一列从a更新到b，那么显示"b" [before="a"]，如果某一列没有被更新，并且更新前的值是c，那么显示"c" [before="c"]

• 对于有主键表的update操作，如果一列从a更新到b，那么显示"b" [before="a"]，如果某一列没有被更新，并且更新前的值是c，那么显示"" [before="c"]

• 解决方案一：例如一张表有三个字段a b c，主键为a，如果update值不更新b，那么b的after就为””，如果想以b的before值作为after的值，可以在源端extract设置TABLE OGG_TEST.*,keycols(a,b);

• 解决方案二：在columnMap中增加isKeyColumn配置，例如：<column src="b" dest="b" destOld="b_old" isKeyColumn="true"/>。

Q： Unsupported major.minor version 52.0

报错信息：

2019-12-23 18:03:58 [main] ERROR UserExitMain - java.lang.UnsupportedClassVersionError: com/aliyun/datahub/client/model/RecordSchema : Unsupported major.minor version 52.0
java.lang.UnsupportedClassVersionError: com/aliyun/datahub/client/model/RecordSchema : Unsupported major.minor version 52.0
    at oracle.goldengate.datasource.DataSourceLauncher.<init>(DataSourceLauncher.java:161)
    at oracle.goldengate.datasource.UserExitMain.main(UserExitMain.java:108)
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Factory method [public final oracle.goldengate.datasource.GGDataSource oracle.goldengate.datasource.factory.DataSourceFactory.getDataSource()] threw exception; nested exception is java.lang.UnsupportedClassVersionError: com/aliyun/datahub/client/model/RecordSchema : Unsupported major.minor version 52.0
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:169)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:570)
    ... 11 more
Caused by: java.lang.UnsupportedClassVersionError: com/aliyun/datahub/client/model/RecordSchema : Unsupported major.minor version 52.0

A: JDK版本不对，插件的编译环境是JDK1.8，用户可能使用了1.7的JDK，建议修改到JDK1.8或者自行去github上clone代码，编译jar包。

Q： Error opening module ./libggjava_ue.so - libjsig.so:cannot open shared object file:no such file or directory

A： 检查版本是否正确，检查文件是否存在，检查环境变量是否生效，重启mgr，重启机器（不推荐，但是之前有用户遇到这个问题，重启了一下机器解决了）。

Q： OGG-00425 No DB login established to retrieve a definition for table xxx.xxx

A: 源端重新生成一下def文件，复制到目标端。

Q： defaultDatahubConfigure is null

报错信息：

2020-01-17 17:28:32 [main] ERROR DatahubHandler - Init error
java.lang.RuntimeException: defaultDatahubConfigure is null
    at com.aliyun.odps.ogg.handler.datahub.ConfigureReader.reader(ConfigureReader.java:117)

A: 插件2.0.3更新了一个配置文件的拼写错误，如果继续使用2.0.2的配置文件，需要修改defalutDatahubConfigure => defaultDatahubConfigure。

Q：目标端报错 OGG-06551 Oracle GoldenGate Collector: Could not translate host name localhost into an Internet address.

A：目标端机器hostname在/etc/hosts里面重新设置localhost对应的ip

Q：找不到jvm相关的so包

A：将jvm的so路径添加到LD_LIBRARY_PATH后，重启mgr。例如：export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:$JAVA_HOME/lib/amd64:$JAVA_HOME/lib/amd64/server

Q：有了DDL语句，比如增加一列，源端ogg没有问题，但是adapter端的ffwriter和jmswriter进程退出，且报错： 2015-06-11 14:01:10 ERROR OGG-01161 Bad column index (5) specified for table OGG_TEST.T_PERSON, max columns = 5.

A：由于表结构改变，需要重做def文件，将重做的def文件放入dirdef后重启即可

Q： OGG report文件中报错 Error (4) retrieving column value.

A: 一般OGG是trail文件解析失败，开启debug日志可以看到具体是哪个RBA报错，调整点位跳过这个RBA方可解决。

六、版本更新说明

2.0.2

• 下载链接

2.0.3

• 下载链接

• 支持采集Oracle表的rowid。

• 支持DataHub的DECIMAL、TIMESTAMP等类型。

• 修改配置参数的拼写错误（defalutDatahubConfigure => defaultDatahubConfigure），如果在升级插件版本之后，依旧使用旧的配置文件，则需要修改此参数。

• 支持压缩、protobuf传输等功能。

2.0.4

• 下载链接

• 支持配置字符集，对于非utf-8编码的源数据库，可以避免中文乱码

2.0.5

• 下载链接

• 性能优化，TPS最高可提升70%

• 日志可配置metric打印

• 修复constColumnMap获取时间与真实时间不一致的bug

2.0.6

• 下载链接

• 修复2.0.5线程数据过多BUG

• 修复after值为空不能哈希的问题

• 支持DataHub新类型，TINYINT ， SMALLINT ， INTEGER ， FLOAT

2.0.7

• 下载链接

• 适配OGG for Bigdata 12.3.2.1，不再支持OGG Adpater

2.0.8

• 下载链接

• 适配OGG for Bigdata 19.1.0.0，不再支持OGG Adpater