AI 助理
文档备案控制台
官方文档
首页 数据总线 DataHub 开发参考 插件说明 OGG插件 OGG for MySQL

OGG for MySQL

更新时间：
复制 MD 格式
产品详情
我的收藏

一、环境要求

  • MySQL 企业版或者社区版

  • OGG for MySQL 12.2.0.2

  • OGG Application Adapter 12.2.0.1

  • java 7（推荐JDK 1.8）

二、安装步骤

以下内容介绍 MySQL 与 OGG 的配置过程，不涉及 MySQL 的安装步骤。示例中的参数配置仅供参考，生产环境的参数请根据实际情况调整。

下载安装包

  • OGG for MySQL

  • OGG Adater

  • DataHub OGG Plugin

    DataHub OGG Plugin

    OGG for MySQL

    OGG Adapter

    支持MySQL版本

    2.0.3

    12.2.0.2

    12.2.0.1

    企业版 / 社区版

    2.0.4

    12.2.0.2

    12.2.0.1

    企业版 / 社区版

    2.0.5

    12.2.0.2

    12.2.0.1

    企业版 / 社区版

OGG for MySQL和OGG Adapter通过OGG官网下载。

OGG 源端配置

  1. 源端MySQL配置

    开启binlog，修改文件/etc/my.cnf，添加如下配置：

    ## for ogg
# Extract expects that all of the table columns are in the binary log. 
# As a result, only binlog_row_image set as full is supported and this is the default
binlog_row_image=full
# It must be set to the value of ROW, which directs the database to log DML statements in binary format.
binlog_format=row
# prevents DDL logging history table entries in the binlog
binlog-ignore-db=oggddl
# binlog path
log-bin=/usr/local/mysql/logs/log
# binlog index path
log-bin-index=/usr/local/mysql/logs/binlog.index
# enable binlog must set server-id, you can modify if there are multiple services
server-id=1

    修改完成后重启MySQL服务。然后创建DB和user

    mysql连接MySQL服务后，输入以下命令创建DB：

    # 创建ogg测试用的数据库ggtt
create database ggtt;

## 创建ogg测试用的用户ogg_test
grant all privileges on ggtt.* to 'ogg_test'@'localhost' identified by 'ogg_test';
flush privileges;

  2. 源端OGG安装

    解压OGG for MySQL压缩文件。

    启动ggsci客户端 ./ggsci，输入命令，创建必需目录。如下命令如果未做特殊说明，均在ggsci客户端中运行

    create subdirs

    运行成功之后便可看到，OGG主目录下增加了几个dirxxx的目录

  3. 配置源端mgr

    编辑mgr配置edit params mgr，配置内容如下

    port 7839
dynamicportlist  7840-7849
purgeoldextracts /home/goldengate/dirdat/*,usecheckpoints, minkeepdays 2

    配置完成后通过命令view params mgr确认配置，再启动mgrstart mgr，查看mgr状态info mgr。

  4. 配置源端extract

    编辑配置 edit params dhext，这里dhext为进程名，可以任意取，和配置中保持一致即可。

    extract dhext
sourcedb ggtt@localhost:3306 userid ogg_test password ogg_test
exttrail dirdat/ms
TranLogOptions AltLogDest /usr/local/mysql/logs/binlog.index
table ggtt.*;

    编辑完成之后保存，添加并启动extact进程

    # 添加进程
add extract dhext,tranlog,begin now

# 每个队列文件大小为200MB
add exttrail ./dirdat/ms,extract dhext, megabytes 200

start dhext

  5. 配置源端pump

    编辑配置 edit params pump

    extract pump
passthru
-- enable compress
rmthost  localhost,mgrport 7839,compress
rmttrail  dirdat/ms
table ggtt.*;

    编辑完成之后保存，添加pump进程

    # 添加进程
add extract pump,exttrailsource dirdat/ms

# 添加目标端的trail文件，每个队列文件大小为200MB
add rmttrail dirdat/ms,extract pump,megabytes 200

# 这里启动pump进程时会连接目标端，因此需要首先启动目标端mgr，否则会报错
start pump

配置目标端

  1. 目标端OGG 安装

    目标端使用 OGG Adapter，无需安装，解压后即可使用。解压后启动 ggsci，执行 create subdirs 命令创建所需目录。执行成功后，OGG 目录下会生成若干以 dir 为前缀的子目录（如 dirprm、dirrpt 等）。

  2. DataHub插件安装和配置

    依赖环境：JDK >= 1.7。配置好JAVA_HOME, LD_LIBRARY_PATH，可以将环境变量配置到~/.bash_profile中，例如：

    export JAVA_HOME=/xxx/xxx
export JRE_HOME=/xxx/xxx/jrexx
export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:$JRE_HOME/lib/amd64:$JRE_HOME/lib/amd64/server

    修改环境变量后，解压datahub-ogg-plugin.tar.gz，修改conf路径下的javaue.properties文件，将{YOUR_HOME}替换为解压后的绝对路径：

    gg.handlerlist=ggdatahub
gg.handler.ggdatahub.type=com.aliyun.odps.ogg.handler.datahub.DatahubHandler
gg.handler.ggdatahub.configureFileName={YOUR_HOME}/aliyun-datahub-ogg-plugin/conf/configure.xml
goldengate.userexit.nochkpt=false
goldengate.userexit.timestamp=utc+8
gg.classpath={YOUR_HOME}/aliyun-datahub-ogg-plugin/lib/*
gg.log.level=debug
jvm.bootoptions=-Xmx512m -Dlog4j.configuration=file:{YOUR_HOME}/datahub-ogg-plugin/conf/log4j.properties -Djava.class.path=ggjava/ggjava.jar

    修改conf路径下的log4j.properties文件，将{YOUR_HOME}替换为解压后的路径：

    log4j.rootLogger=INFO,fileout
log4j.logger.com.goldengate=DEBUG
log4j.logger.com.aliyun=DEBUG

# Console output...
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

# Rolling log file output...
log4j.appender.fileout=org.apache.log4j.RollingFileAppender
log4j.appender.fileout.File={YOUR_HOME}/aliyun-datahub-ogg-plugin/log/datahub-ogg-plugin.log
log4j.appender.fileout.MaxFileSize=10000KB
log4j.appender.fileout.MaxBackupIndex=30
log4j.appender.fileout.layout=org.apache.log4j.PatternLayout
log4j.appender.fileout.layout.conversionPattern=%d{yyyy-MM-dd HH:mm:ss} [%t] %-5p %c{1} - %m%n

    修改conf路径下的configure.xml文件，修改方式见文件中的注释。

    需要重点关注的参数 

    • sid任意填写

    • schema和oracleSchema填写MySQL的dbname，在本测试用例中填写ggtt

    • oracleTable填写MySQL 的table

    • rowIdColumn不能填写，因为MySQL无法获取rowid

    <?xml version="1.0" encoding="UTF-8"?>
<configue>
    <defaultOracleConfigure>
        <!-- 任意填写 必选-->
        <sid>100</sid>
        <!-- mysql database, 可以被mapping中的oracleSchema覆盖, 两者必须有一个非空-->
        <schema>ogg_test</schema>
    </defaultOracleConfigure>
    <defalutDatahubConfigure>
        <!-- datahub endpoint, 必填-->
        <endPoint>YOUR_DATAHUB_ENDPOINT</endPoint>
        <!-- datahub project, 可以被mapping中的datahubProject, 两者必须有一个非空-->
        <project>YOUR_DATAHUB_PROJECT</project>
        <!-- datahub accessId, 可以被mapping中的datahubAccessId覆盖, 两者必须有一个非空-->
        <accessId>YOUR_DATAHUB_ACCESS_ID</accessId>
        <!-- datahub accessKey, 可以被mapping中的datahubAccessKey覆盖, 两者必须有一个非空-->
        <accessKey>YOUR_DATAHUB_ACCESS_KEY</accessKey>
        <!-- 数据变更类型同步到datahub对应的字段，可以被columnMapping中的ctypeColumn覆盖 -->
        <ctypeColumn>optype</ctypeColumn>
        <!-- 数据变更时间同步到datahub对应的字段，可以被columnMapping中的ctimeColumn覆盖 -->
        <ctimeColumn>readtime</ctimeColumn>
        <!-- 数据变更序号同步到datahub对应的字段, 按数据变更先后递增, 不保证连续, 可以被columnMapping中的cidColumn覆盖 -->
        <cidColumn>record_id</cidColumn>
    </defalutDatahubConfigure>
    <!-- 默认最严格，不落文件 直接退出 无限重试-->
    <!-- 运行每批上次的最多纪录数, 可选, 默认1000-->
    <batchSize>1000</batchSize>
    <!-- 默认时间字段转换格式, 可选, 默认yyyy-MM-dd HH:mm:ss-->
    <defaultDateFormat>yyyy-MM-dd HH:mm:ss</defaultDateFormat>
    <!-- 脏数据是否继续, 可选, 默认false-->
    <dirtyDataContinue>true</dirtyDataContinue>
    <!-- 脏数据文件, 可选, 默认datahub_ogg_plugin.dirty-->
    <dirtyDataFile>datahub_ogg_plugin.dirty</dirtyDataFile>
    <!-- 脏数据文件最大size, 单位M, 可选, 默认500-->
    <dirtyDataFileMaxSize>200</dirtyDataFileMaxSize>
    <!-- 重试次数, -1:无限重试 0:不重试 n:重试次数, 可选, 默认-1-->
    <retryTimes>0</retryTimes>
    <!--指定shard id列表, 优先生效, 可选,例如0,1-->
    <shardId>0,1</shardId>
    <!-- 重试间隔, 单位毫秒, 可选, 默认3000-->
    <retryInterval>4000</retryInterval>
    <!-- 点位文件, 可选, 默认datahub_ogg_plugin.chk-->
    <checkPointFileName>datahub_ogg_plugin.chk</checkPointFileName>
    <mappings>
        <mapping>
            <!-- mysql database, 见上描述-->
            <oracleSchema></oracleSchema>
            <!-- mysql table, 必选-->
            <oracleTable>t_person</oracleTable>
            <!-- datahub project, 见上描述-->
            <datahubProject></datahubProject>
            <!-- datahub topic, 必选-->
            <datahubTopic>t_person</datahubTopic>
            <ctypeColumn></ctypeColumn>
            <ctimeColumn></ctimeColumn>
            <cidColumn></cidColumn>
            <columnMapping>
                <!--
                src:oracle字段名称, 必须;
                dest:datahub field, 必须;
                destOld:变更前数据落到datahub的field, 可选;
                isShardColumn: 是否作为shard的hashkey, 可选, 默认为false, 可以被shardId覆盖
                isDateFormat: timestamp字段是否采用DateFormat格式转换, 默认true. 如果是false, 源端数据必须是long
                dateFormat: timestamp字段的转换格式, 不填就用默认值
                -->
                <column src="id" dest="id" isShardColumn="true"  isDateFormat="false" dateFormat="yyyy-MM-dd HH:mm:ss"/>
                <column src="name" dest="name" isShardColumn="true"/>
                <column src="age" dest="age"/>
                <column src="address" dest="address"/>
                <column src="comments" dest="comments"/>
                <column src="sex" dest="sex"/>
                <column src="temp" dest="temp" destOld="temp1"/>
            </columnMapping>
        </mapping>
    </mappings>
</configue>

  3. 配置目标端mgr

    编辑mgr配置edit params mgr，然后启动mgr start mgr

    port 7839
dynamicportlist 7840-7849
purgeoldextracts dirdat/*,usecheckpoints, minkeepdays 7

  4. 配置目标端writer

    编辑配置edit params dhwriter

    extract dhwriter
getEnv (JAVA_HOME)
getEnv (LD_LIBRARY_PATH)
getEnv (PATH)
CUSEREXIT libggjava_ue.so CUSEREXIT PASSTHRU INCLUDEUPDATEBEFORES, PARAMS "{YOUR_HOME}/datahub-ogg-plugin/conf/javaue.properties"
GetUpdateBefores
table ggtt.*;

    编辑完成之后保存，添加extact进程并启动

    # 添加进程
add extract dhwriter, exttrailsource dirdat/ms

# 启动
start dhwriter

三、MySQL和DataHub字段类型对应

DataHub的TIMESTAMP类型存储微秒时间戳，字段映射到TIMESTAMP类型时，都会被转为微秒时间戳。

MySQL

DataHub

备注

CHAR、VARCHAR

STRING

INT、TINYINT、BIG INT

BIGINT

DECIMAL、FLOAT、DOUBLE

DOUBLE/DECIMAL

根据精度选择

DATETIME、TIMESTAMP

TIMESTAMP

TEXT、TINYTEXT

STRING

以上是MySQL常见的数据类型和DataHub的对应，更多MySQL支持的数据类型可以参考官方文档。

上一篇：OGG for Big Data下一篇：OGG for Bigdata(Kafka)