[toc]

前言(为了安装oraclegoldgate for oracle准备)

通过X11实现 Linux服务器图形化界面显示

1 背景描述

有些LINUX服务器出于性能和效率的考虑，通常都是没有安装图形化界面的，那么图形化程序在服务器上压根儿就跑不起来，或者无法直接显示出来，这就很尴尬了！那么如何解决这个问题呢？可以基于X11 Forwarding技术 + MobaXterm 工具，就可以轻松搞定，似不似很简单？

2 原理介绍

2.1 X协议

Linux 本身是没有图形化界面的，所谓的图形化界面系统只不过中 Linux 下的应用程序。这一点和 Windows 不一样。Windows 从 Windows 95 开始，图形界面就直接在系统内核中实现了，是操作系统不可或缺的一部分。Linux 的图形化界面，底层都是基于 X 协议。
X 协议由 X server 和 X client 组成：

image.png

l X server 管理主机上与显示相关的硬件设置（如显卡、硬盘、鼠标等），它负责屏幕画面的绘制与显示，以及将输入设置（如键盘、鼠标）的动作告知 X client。

l X client (即 X 应用程序) 则主要负责事件的处理（即程序的逻辑）。

举个例子，如果用户点击了鼠标左键，因为鼠标归 X server 管理，于是 X server 就捕捉到了鼠标点击这个动作，然后它将这个动作告诉 X client，因为 X client 负责程序逻辑，于是 X client 就根据程序预先设定的逻辑（例如画一个圆），告诉 X server 说：“请在鼠标点击的位置，画一个圆”。最后，X server 就响应 X client 的请求，在鼠标点击的位置，绘制并显示出一个圆。

2.2 X11 Forwarding

image.png

这么绕，有啥意义呢？当然有！
许多时候 X server 和 X client 在同一台主机上，这看起来没什么。但是， X server 和 X client 完全可以运行在不同的机器上，只要彼此通过 X 协议通信即可。于是，我们就可以做一些“神奇”的事情，比如像本文开头谈到的，在本地显示 (X server)，运行在服务器上的 GUI 程序 (X client)。这样的操作可以通过 SSH X11 Forwarding (转发) 来实现。
X11 中的 X 指的就是 X 协议，11 指的是采用 X 协议的第 11 个版本。

2.3 MobaXterm

image.png

那 MobaXterm 又是什么鬼？MobaXterm 是一款开源、免费的、全功能终端软件。它与 PuTTY 类似，但却比 PuTTY 要强大得多，其中一个很实用的功能就是 MobaXterm 自带 X Server。这样我们就不用劳神地去想怎么在 Windows 上启动 X server 了。

小结一下，整个实现逻辑就是：
本地机器采用 MobaXterm (自带 X server) 连接远程服务器。然后，在服务器上运行 GUI 程序 (即 X client)，通过 SSH X11 Forwarding，转发到本地 (Windows 机器上)。

3 实战介绍

3.1 服务器端（LINUX7.x）

3.1.1 修改SSH配置

[root@linux-template-7 ~]# cat /etc/ssh/sshd_config

X11Forwarding yes

X11UseLocalhost no -- 禁止将X11转发请求绑定到本地回环地址上

AddressFamily inet -- 强制使用IPv4通道

配置不对请修改成上面这样

3.1.2 重启SSH服务

[root@linux-template-7 ~]# systemctl restart sshd.service

3.2 客户端（WIN7）

3.2.1 安装MobaXterm

https://mobaxterm.mobatek.net/download.html

image.png

3.2.2 打开本地终端

image.png

3.2.3 SSH连接服务器

image.png

3.2.4 执行xclock程序

已经弹出xclock图形界面了，似不似很简单，似不似很Easy，那就一起实践下吧：）

image.png

---

正文

ogg即Oracle GoldenGate是Oracle的同步工具，本文讲如何配置ogg以实现Oracle数据库增量数据实时同步到kafka中，其中同步消息格式为json。
下面是我的源端和目标端的一些配置信息：

• | 版本 | OGG版本 | ip | 别名
  ---|--- |--- |--- |---
  源端| Oracle 12c |Oracle GoldenGate 19.1.0.0 | 192.168.8.22| oracle|
  目标端| kafka_2.12-2.3.0 |Oracle GoldenGate for Big Data 19.1.0.0 | 192.168.8.31| bigdata |

1、下载

可在这里或旧版本查询下载
注意：源端和目标端的文件不一样，目标端需要下载Oracle GoldenGate for Big Data,源端需要下载Oracle GoldenGate for Oracle具体下载方法见最后的附录截图。

2、源端（Oracle）配置

注意：源端是安装了oracle的机器，oracle环境变量之前都配置好了

2.1 解压

先建立ogg目录

mkdir -p /opt/ogg
unzip V983320-01.zip

解压后得到一个tar包，再解压这个tar

tar xf fbo_ggs_Linux_x64_shiphome.tar -C /opt/ogg
chown -R oracle:oinstall /opt/ogg

（使oracle用户有ogg的权限，后面有些需要在oracle用户下执行才能成功）

2.2 配置ogg环境变量

为了简单方便起见，我在/etc/profile里配置的，建议在生产中配置oracle的环境变量文件/home/oracle/.bash_profile里配置，为了怕出问题，我把OGG_HOME等环境变量在/etc/profile配置了一份，不知道这是否是必须的。

vim /etc/profile

export OGG_HOME=/opt/ogg
export LD_LIBRARY_PATH=$ORACLE_HOME/lib:/usr/lib
export PATH=$OGG_HOME:$PATH

使之生效

source /etc/profile

测试一下ogg命令

cd /opt/ogg
./runInstaller

按照上面得安装以后会本mobaxtern弹出一个oracle安装对话框

image.png

点击下一步

image.png image.png image.png image.png

如果命令成功即可进行下一步，不成功请检查前面的步骤

2.3 oracle打开归档模式

su - oracle
sqlplus / as sysdba

执行下面的命令查看当前是否为归档模式

archive log list

SQL> archive log list 
Database log mode          No Archive Mode
Automatic archival         Disabled
Archive destination        USE_DB_RECOVERY_FILE_DEST
Oldest online log sequence     12
Current log sequence           14

若为Disabled，手动打开即可

conn / as sysdba (以DBA身份连接数据库) 
shutdown immediate (立即关闭数据库)
startup mount (启动实例并加载数据库，但不打开)
alter database archivelog; (更改数据库为归档模式)
alter database open; (打开数据库)
alter system archive log start; (启用自动归档)

再执行一下

archive log list

Database log mode          Archive Mode
Automatic archival         Enabled
Archive destination        USE_DB_RECOVERY_FILE_DEST
Oldest online log sequence     12
Next log sequence to archive   14
Current log sequence           14

可以看到为Enabled，则成功打开归档模式。

2.4 Oracle打开日志相关

OGG基于辅助日志等进行实时传输，故需要打开相关日志确保可获取事务内容，通过下面的命令查看该状态

select force_logging, supplemental_log_data_min from v$database;

FORCE_ SUPPLEMENTAL_LOG
------ ----------------
NO     NO

若为NO，则需要通过命令修改

alter database force logging;
alter database add supplemental log data;
ALTER SYSTEM SET ENABLE_GOLDENGATE_REPLICATION = TRUE;#设置允许GoldDate读取日志

再查看一下为YES即可

SQL> select force_logging, supplemental_log_data_min from v$database;

FORCE_ SUPPLEMENTAL_LOG
------ ----------------
YES    YES

2.5 oracle创建复制用户

首先root用户建立相关文件夹，并赋予权限

mkdir -p /u01/app/oracle/oggdata/orcl
chown -R oracle:oinstall /u01/app/oracle/oggdata/orcl

然后执行下面sql

SQL> create tablespace oggtbs datafile '/u01/app/oracle/oggdata/orcl/oggtbs01.dbf' size 1000M autoextend on;

Tablespace created.

SQL>  create user ogg identified by ogg default tablespace oggtbs;

User created.

SQL> grant dba to ogg;

Grant succeeded.

2.6 OGG初始化

ggsci
create subdirs

ggsci

Oracle GoldenGate Command Interpreter for Oracle
Version 19.1.0.0.2 OGGCORE_19.1.0.0.0_PLATFORMS_190823.0013_FBO
Linux, x64, 64bit (optimized), Oracle 12c on Aug 23 2019 20:05:42
Operating system character set identified as UTF-8.

Copyright (C) 1995, 2019, Oracle and/or its affiliates. All rights reserved.



GGSCI (crmtstmicserv) 1> create subdirs

Creating subdirectories under current directory /root

Parameter files                /root/dirprm: created
Report files                   /root/dirrpt: created
Checkpoint files               /root/dirchk: created
Process status files           /root/dirpcs: created
SQL script files               /root/dirsql: created
Database definitions files     /root/dirdef: created
Extract data files             /root/dirdat: created
Temporary files                /root/dirtmp: created
Stdout files                   /root/dirout: created


GGSCI (crmtstmicserv) 2>

2.7 Oracle创建测试表

创建一个用户,在该用户下新建测试表，用户名、密码、表名均为 test_ogg。

create user test_ogg  identified by test_ogg default tablespace users;
grant dba to test_ogg;
conn test_ogg/test_ogg;
create table test_ogg(id int ,name varchar(20),primary key(id));

3 目标端（kafka）配置

3.1 解压

mkdir -p /opt/ogg
unzip 123111_ggs_Adapters_Linux_x64.zip 
tar xf ggs_Adapters_Linux_x64.tar  -C /opt/ogg/

3.2 环境变量

vim /etc/profile

export OGG_HOME=/opt/ogg
export LD_LIBRARY_PATH=$JAVA_HOME/jre/lib/amd64:$JAVA_HOME/jre/lib/amd64/server:$JAVA_HOME/jre/lib/amd64/libjsig.so:$JAVA_HOME/jre/lib/amd64/server/libjvm.so:$OGG_HOME/lib
export PATH=$OGG_HOME:$PATH

source /etc/profile

同样测试一下ogg命令

ggsci

3.3 初始化目录

create subdirs

4、OGG源端配置

4.1 配置OGG的全局变量

先切换到oracle用户下

su oracle
cd /opt/ogg
ggsci

GGSCI (crmtstmicserv) 1> dblogin userid ogg password ogg
Successfully logged into database.

GGSCI (crmtstmicserv) 2> edit param ./globals

然后和用vim编辑一样添加

oggschema ogg

4.2 配置管理器mgr

GGSCI (crmtstmicserv) 3> edit param mgr
PORT 7809
DYNAMICPORTLIST 7810-7909
AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

说明：PORT即mgr的默认监听端口；DYNAMICPORTLIST动态端口列表，当指定的mgr端口不可用时，会在这个端口列表中选择一个，最大指定范围为256个；AUTORESTART重启参数设置表示重启所有EXTRACT进程，最多5次，每次间隔3分钟；PURGEOLDEXTRACTS即TRAIL文件的定期清理

4.3 添加复制表或者SCHEMA

GGSCI (crmtstmicserv) 4> add trandata test_ogg.test_ogg

Logging of supplemental redo data enabled for table TEST_OGG.TEST_OGG.

GGSCI (crmtstmicserv) 5> info trandata test_ogg.test_ogg

Logging of supplemental redo log data is enabled for table TEST_OGG.TEST_OGG.

Columns supplementally logged for table TEST_OGG.TEST_OGG: ID

如果添加按照schema 如下配置:

GGSCI (crmtstmicserv) 4> add schematrandata crm_report ALLCOLS #crm_report为我的schema名

Logging of supplemental redo data enabled for table TEST_OGG.TEST_OGG.

GGSCI (crmtstmicserv) 5> info schematrandata crm_report

4.4 配置extract进程

GGSCI (crmtstmicserv) 6> edit param extkafka
extract extkafka
dynamicresolution
SETENV (ORACLE_SID = "orcl")
SETENV (NLS_LANG = "american_america.AL32UTF8")
userid ogg,password ogg
exttrail /opt/ogg/dirdat/to
table test_ogg.test_ogg;

schema的话把table 这一行换成:

table crm_report.*;

说明：第一行指定extract进程名称；dynamicresolution动态解析；SETENV设置环境变量，这里分别设置了Oracle数据库以及字符集；userid ggs,password ggs即OGG连接Oracle数据库的帐号密码，这里使用2.5中特意创建的复制帐号；exttrail定义trail文件的保存位置以及文件名，注意这里文件名只能是2个字母，其余部分OGG会补齐；table即复制表的表名，支持*通配，必须以;结尾

添加extract进程：

GGSCI (crmtstmicserv) 16> add extract extkafka,tranlog,begin now
EXTRACT added.

(注：若报错

ERROR: Could not create checkpoint file /opt/ogg/dirchk/EXTKAFKA.cpe (error 2, No such file or directory).

执行下面的命令再重新添加即可。

create subdirs

添加trail文件的定义与extract进程绑定：

GGSCI (crmtstmicserv) 17> add exttrail /opt/ogg/dirdat/to,extract extkafka
EXTTRAIL added.

4.5 配置pump进程

pump进程本质上来说也是一个extract，只不过他的作用仅仅是把trail文件传递到目标端，配置过程和extract进程类似，只是逻辑上称之为pump进程

GGSCI (crmtstmicserv) 18> edit param pukafka
extract pukafka
passthru
dynamicresolution
userid ogg,password ogg
rmthost 192.168.44.129 mgrport 7809
rmttrail /opt/ogg/dirdat/to
table test_ogg.test_ogg;

schema的话把table 这一行换成:

table crm_report.*;

说明：第一行指定extract进程名称；passthru即禁止OGG与Oracle交互，我们这里使用pump逻辑传输，故禁止即可；dynamicresolution动态解析；userid ogg,password ogg即OGG连接Oracle数据库的帐号密码rmthost和mgrhost即目标端(kafka)OGG的mgr服务的地址以及监听端口；rmttrail即目标端trail文件存储位置以及名称。

分别将本地trail文件和目标端的trail文件绑定到extract进程：

GGSCI (crmtstmicserv) 1> add extract pukafka,exttrailsource /opt/ogg/dirdat/to
EXTRACT added.
GGSCI (crmtstmicserv) 2> add rmttrail /opt/ogg/dirdat/to,extract pukafka
RMTTRAIL added.

4.6 配置define文件

Oracle与MySQL，Hadoop集群（HDFS，Hive，kafka等）等之间数据传输可以定义为异构数据类型的传输，故需要定义表之间的关系映射，在OGG命令行执行：

GGSCI (crmtstmicserv) 3> edit param test_ogg
defsfile /opt/ogg/dirdef/test_ogg.test_ogg
userid ogg,password ogg
table test_ogg.test_ogg;

schema的话把table 这一行换成:

table crm_report.*;

在OGG主目录下执行(oracle用户)：

./defgen paramfile dirprm/crm_report.prm

***********************************************************************
        Oracle GoldenGate Table Definition Generator for Oracle
 Version 11.2.1.0.3 14400833 OGGCORE_11.2.1.0.3_PLATFORMS_120823.1258
   Linux, x64, 64bit (optimized), Oracle 11g on Aug 23 2012 16:58:29
 
Copyright (C) 1995, 2012, Oracle and/or its affiliates. All rights reserved.


                    Starting at 2018-05-23 05:03:04
***********************************************************************

Operating System Version:
Linux
Version #1 SMP Wed Apr 12 15:04:24 UTC 2017, Release 3.10.0-514.16.1.el7.x86_64
Node: crmtstmicserv
Machine: x86_64
                         soft limit   hard limit
Address Space Size   :    unlimited    unlimited
Heap Size            :    unlimited    unlimited
File Size            :    unlimited    unlimited
CPU Time             :    unlimited    unlimited

Process id: 13126

***********************************************************************
**            Running with the following parameters                  **
***********************************************************************
defsfile /opt/ogg/dirdef/crm_report
userid ogg,password ***
table crm_report.*;
Retrieving definition for TEST_OGG.TEST_OGG



Definitions generated for 1 table in /opt/ogg/dirdef/crm_report

将生成的/opt/ogg/dirdef/crm_report发送的目标端ogg目录下的dirdef里：

scp -r /opt/ogg/dirdef/crm_report root@slave1:/opt/ogg/dirdef/

5、OGG目标端配置

5.1 开启kafka服务

cd /opt/kafka_2.11-1.1.0/
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

5.2 配置管理器mgr

GGSCI (bigdata) 1>  edit param mgr
PORT 7809
DYNAMICPORTLIST 7810-7909
AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

5.3 配置checkpoint

checkpoint即复制可追溯的一个偏移量记录，在全局配置里添加checkpoint表即可。

edit  param  ./GLOBALS
CHECKPOINTTABLE crm_report.checkpoint

5.4 配置replicate进程

GGSCI (bigdata) 4> edit param rekafka
REPLICAT rekafka
sourcedefs /opt/ogg/dirdef/test_ogg.test_ogg
TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
REPORTCOUNT EVERY 1 MINUTES, RATE 
GROUPTRANSOPS 10000
MAP test_ogg.test_ogg, TARGET test_ogg.test_ogg;

说明：REPLICATE rekafka定义rep进程名称；sourcedefs即在4.6中在源服务器上做的表映射文件；TARGETDB LIBFILE即定义kafka一些适配性的库文件以及配置文件，配置文件位于OGG主目录下的dirprm/kafka.props；REPORTCOUNT即复制任务的报告生成频率；GROUPTRANSOPS为以事务传输时，事务合并的单位，减少IO操作；MAP即源端与目标端的映射关系

5.5 配置kafka.props

cd /opt/ogg/dirprm/
vim kafka.props

gg.handlerlist=kafkahandler //handler类型
gg.handler.kafkahandler.type=kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties //kafka相关配置
gg.handler.kafkahandler.topicMappingTemplate=crm_report_ogg //kafka的topic名称，无需手动创建
gg.handler.kafkahandler.format=json //传输文件的格式，支持json，xml等
gg.handler.kafkahandler.mode=op  //OGG for Big Data中传输模式，即op为一次SQL传输一次，tx为一次事务传输一次
gg.classpath=dirprm/:/opt/kafka_2.11-1.1.0/libs/*:/opt/ogg/:/opt/ogg/lib

vim custom_kafka_producer.properties
bootstrap.servers=localhost:9092 //kafkabroker的地址
acks=1
compression.type=gzip //压缩类型
reconnect.backoff.ms=1000 //重连延时
value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
batch.size=102400
linger.ms=10000

其中需要将后面的注释去掉，ogg不识别注释，如果不去掉会报错

5.6 添加trail文件到replicate进程

GGSCI (bigdata) 2> add replicat rekafka exttrail /opt/ogg/dirdat/to,checkpointtable test_ogg.checkpoint
REPLICAT added.

6、测试

6.1 启动所有进程

在源端和目标端的OGG命令行下使用start [进程名]的形式启动所有进程。
启动顺序按照源mgr——目标mgr——源extract——源pump——目标replicate来完成。
全部需要在ogg目录下执行ggsci目录进入ogg命令行。
源端依次是

start mgr
start extkafka
start pukafka

目标端

start mgr
start rekafka

可以通过info all 或者info [进程名] 查看状态，所有的进程都为RUNNING才算成功
源端

GGSCI (crmtstmicserv) 5> info all

Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING                                           
EXTRACT     RUNNING     EXTKAFKA    04:50:21      00:00:03    
EXTRACT     RUNNING     PUKAFKA     00:00:00      00:00:03

目标端

GGSCI (bigdata) 3> info all

Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING                                           
REPLICAT    RUNNING     REKAFKA     00:00:00      00:00:01

6.2 异常解决

如果有不是RUNNING可通过查看日志的方法检查解决问题，具体通过下面两种方法

vim ggser.log

或者ogg命令行,以rekafka进程为例

GGSCI (bigdata) 2> view report rekafka

列举其中我遇到的一个问题：
异常信息

SEVERE: Unable to set property on handler 'kafkahandler' (oracle.goldengate.handler.kafka.KafkaHandler). Failed to set property: TopicName:="test_ogg" (class: oracle.goldengate.handler.kafka.KafkaHandler).
oracle.goldengate.util.ConfigException: Failed to set property: TopicName:="test_ogg" (class: oracle.goldengate.handler.kafka.KafkaHandler).
at ......

具体原因是网上的教程是旧版的，设置topicName的属性为:

gg.handler.kafkahandler.topicName=test_ogg

新版的这样设置

gg.handler.kafkahandler.topicMappingTemplate=test_ogg

大家可根据自己的版本进行设置，附上stackoverflow原答案

I tried to move data from Oracle Database to Kafka using Golden gate adapter Version 12.3.0.1.0

In new version there is no topicname

The following resolves the topic name using the short table name
gg.handler.kafkahandler.topicMappingTemplate=test

In previous version we have gg.handler.kafkahandler.topicName=test

6.3 测试同步更新效果

现在源端执行sql语句

conn test_ogg/test_ogg
insert into test_ogg values(1,'test');
commit;
update test_ogg set name='zhangsan' where id=1;
commit;
delete test_ogg where id=1;
commit;

查看源端trail文件状态

ls -l /opt/ogg/dirdat/to*
-rw-rw-rw- 1 oracle oinstall 1464 May 23 10:31 /opt/ogg/dirdat/to000000

查看目标端trail文件状态

ls -l /opt/ogg/dirdat/to*
-rw-r----- 1 root root 1504 May 23 10:31 /opt/ogg/dirdat/to000000

查看kafka是否自动建立对应的主题

bin/kafka-topics.sh --list --zookeeper localhost:2181

在列表中显示有test_ogg则表示没问题
通过消费者看是否有同步消息

bin/kafka-console-consumer.sh --bootstrap-server 192.168.44.129:9092 --topic test_ogg --from-beginning

{"table":"TEST_OGG.TEST_OGG","op_type":"I","op_ts":"2018-05-23 10:31:28.000078","current_ts":"2018-05-23T10:36:48.525000","pos":"00000000000000001093","after":{"ID":1,"NAME":"test"}}
{"table":"TEST_OGG.TEST_OGG","op_type":"U","op_ts":"2018-05-23 10:31:36.000073","current_ts":"2018-05-23T10:36:48.874000","pos":"00000000000000001233","before":{},"after":{"ID":1,"NAME":"zhangsan"}}
{"table":"TEST_OGG.TEST_OGG","op_type":"D","op_ts":"2018-05-23 10:31:43.000107","current_ts":"2018-05-23T10:36:48.875000","pos":"00000000000000001376","before":{"ID":1}}

显然，Oracle的数据已准实时同步到Kafka,格式为json,其中op_type代表操作类型，这个可配置，我没有配置则按默认的来，默认为

gg.handler.kafkahandler.format.insertOpKey = I  
gg.handler.kafkahandler.format.updateOpKey = U  
gg.handler.kafkahandler.format.deleteOpKey = D

before代表操作之前的数据，after代表操作后的数据，现在已经可以从kafka获取到同步的json数据了，后面可以用SparkStreaming和Storm等解析然后存到hadoop等大数据平台里

6.4 SparkStreaming测试消费同步消息
具体代码可参考Spark Streaming连接Kafka入门教程
下面附上消费成功的结果图

image.png

7、更新：后续遇到的问题

在后面的使用过程中发现上面同步到kafka的json数据中少一些我们想要的一些，下面讲一下我是如何解决的
首先建表：

CREATE TABLE "TCLOUD"."T_OGG2" 
   (    "ID" NUMBER(*,0), 
    "TEXT_NAME" VARCHAR2(20), 
    "AGE" NUMBER(*,0), 
    "ADD" VARCHAR2(100), 
    "IDD" VARCHAR2(100), 
     CONSTRAINT "T_OGG2_PK" PRIMARY KEY ("ID", "IDD")

   )

为什么不用之前建的表，主要是之前的字段太少，不容易看出问题，现在主要是增加几个字段，然后id,idd是联合主键。
看一下按照之前的配置，同步到kafka的数据(截取部分数据)

{"table":"TCLOUD.T_OGG2","op_type":"I","op_ts":"2018-05-31 11:46:09.512672","current_ts":"2018-05-31T11:46:15.292000","pos":"00000000000000001903","after":{"ID":4,"TEXT_NAME":null,"AGE":0,"ADD":null,"IDD":"8"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 11:49:10.514549","current_ts":"2018-05-31T11:49:16.450000","pos":"00000000000000002227","before":{},"after":{"ID":4,"TEXT_NAME":"lisi","IDD":"7"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 11:49:48.514869","current_ts":"2018-05-31T11:49:54.481000","pos":"00000000000000002373","before":{"ID":4,"IDD":"7"},"after":{"ID":1,"IDD":"7"}}

{"table":"TCLOUD.T_OGG2","op_type":"D","op_ts":"2018-05-31 11:52:38.516877","current_ts":"2018-05-31T11:52:45.633000","pos":"00000000000000003161","before":{"ID":1,"IDD":"7"}}

现在只有insert的数据是全的，update更新非主键字段before是没有数据的，更新主键before只有主键的数据，delete只有before的主键字段，也就是update和delete的信息是不全的，且没有主键信息（程序里是不能判断哪一个是主键的），这样对于程序自动解析同步数据是不利的（不同的需求可能不一样），具体自己可以分析，就不啰嗦了，这里主要解决，有需要before和after全部信息和主键信息的需求。

7.1 添加before

在源端extract里添加下面几行

GGSCI (crmtstmicserv) 33> edit param extkafka
GETUPDATEBEFORES
NOCOMPRESSDELETES
NOCOMPRESSUPDATES

重启 extkafka

stop extkafka
start extkafka

然后测试

{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 14:48:55.630340","current_ts":"2018-05-31T14:49:01.709000","pos":"00000000000000003770","before":{"ID":1,"AGE":20,"IDD":"1"},"after":{"ID":1,"AGE":1,"IDD":"1"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 14:48:55.630340","current_ts":"2018-05-31T14:49:01.714000","pos":"00000000000000004009","before":{"ID":1,"AGE":20,"IDD":"2"},"after":{"ID":1,"AGE":1,"IDD":"2"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 14:48:55.630340","current_ts":"2018-05-31T14:49:01.715000","pos":"00000000000000004248","before":{"ID":1,"AGE":20,"IDD":"8"},"after":{"ID":1,"AGE":1,"IDD":"8"}}

发现update之后before里有数据即可，但是现在before和after的数据都不全（只有部分字段）

网上有的说只添加GETUPDATES即可，但我测试了没有成功，关于每个配置项什么含义可以参考https://blog.csdn.net/linucle/article/details/13505939（有些配置的含义里面也没有给出）
参考：http://www.itpub.net/thread-2083473-1-1.html

7.2 添加主键

在kafka.props添加

gg.handler.kafkahandler.format.includePrimaryKeys=true

重启 rekafka

stop rekafka
start rekafka

测试：

{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 14:58:57.637035","current_ts":"2018-05-31T14:59:03.401000","pos":"00000000000000004510","primary_keys":["ID","IDD"],"before":{"ID":1,"AGE":1,"IDD":"1"},"after":{"ID":1,"AGE":20,"IDD":"1"}}

发现有primary_keys，不错~

参考：http://blog.51cto.com/lyzbg/2088409

7.3 补全全部字段

如果字段补全应该是Oracle没有开启全列补充日志

SQL> select supplemental_log_data_all from v$database;  

SUPPLE
------
NO

通过以下命令开启

SQL> alter database add supplemental log data(all) columns;

Database altered.

SQL> select supplemental_log_data_all from v$database;

SUPPLE
------
YES

SQL>

测试一下

{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 15:27:45.655518","current_ts":"2018-05-31T15:27:52.891000","pos":"00000000000000006070","primary_keys":["ID","IDD"],"before":{"ID":1,"TEXT_NAME":null,"AGE":1,"ADD":null,"IDD":"1"},"after":{"ID":1,"TEXT_NAME":null,"AGE":20,"ADD":null,"IDD":"1"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 15:27:45.655518","current_ts":"2018-05-31T15:27:52.893000","pos":"00000000000000006341","primary_keys":["ID","IDD"],"before":{"ID":1,"TEXT_NAME":null,"AGE":1,"ADD":null,"IDD":"2"},"after":{"ID":1,"TEXT_NAME":null,"AGE":20,"ADD":null,"IDD":"2"}}
{"table":"TCLOUD.T_OGG2","op_type":"U","op_ts":"2018-05-31 15:27:45.655518","current_ts":"2018-05-31T15:27:52.895000","pos":"00000000000000006612","primary_keys":["ID","IDD"],"before":{"ID":1,"TEXT_NAME":null,"AGE":1,"ADD":null,"IDD":"8"},"after":{"ID":1,"TEXT_NAME":null,"AGE":20,"ADD":null,"IDD":"8"}}

到现在json信息里的内容已经很全了，基本满足了我想要的，附图：

启发我发现和Oracle全列补充日志没有开启有关的博客：https://blog.csdn.net/huoshuyinhua/article/details/79013387
开启命令参考：https://blog.csdn.net/aaron8219/article/details/16825963

注：博客上讲到，开启全列补充日志会导致磁盘快速增长，LGWR进程繁忙，不建议使用。大家可根据自己的情况使用。

8、关于通配

如果想通配整个库的话，只需要把上面的配置所有表名的改为，如test_ogg.test_ogg改为 test_ogg.,但是kafka的topic不能通配，所以需要把所有的表的数据放在一个topic即可，后面再用程序解析表名即可。