AI 助理
备案控制台
开发者社区 数据库 文章 正文

聊聊 datax 的 OceanBase 数据同步插件 ||批处理参数 rewriteBatchedStatements=true&useCursorFetch=true

2023-09-22 276
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
RDS MySQL Serverless 基础系列，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
云数据库 RDS MySQL，集群系列 2核4GB
推荐场景：
搭建个人博客
云数据库 RDS MySQL，高可用系列 2核4GB
简介： 聊聊 datax 的 OceanBase 数据同步插件分析下批处理参数 rewriteBatchedStatements=true&useCursorFetch=true 对大规模数据读写的性能影响

聊聊 datax 的 OceanBase 数据同步插件 ||批处理参数 rewriteBatchedStatements=true&useCursorFetch=true

1 背景

  • 在信创的大背景下，不少公司选用了蚂蚁的分布式数据库 OceanBase，OceanBase 是一款开源分布式 HTAP（Hybrid Transactional/Analytical Processing）数据库管理系统，具有原生分布式架构，支持金融级高可用、透明水平扩展、分布式事务、多租户和语法兼容等企业级特性。OceanBase 内核通过大规模商用场景的考验，已服务众多行业客户，现面向未来持续构建内核技术竞争力。

  • 在大数据场景下，不可避免地会遇到使用数据同步工具在 hdfs 和 OceanBase 之间同步数据的需求，常见的离线数据同步工具有 sqoop/datax/spark/seatunnel等。

  • 在使用 datax 同步OceanBase数据时，我们可以使用rdbmswriter/rdbmsreader,也可以使用oceanbasev10writer/oceanbasev10reader来同步 ob数据。

近期我们在某客户现场使用datax 的 rdbmswriter/rdbmsreader 进行大批量数据同步时却遇到了 OutOfMemoryError 问题，本文针对该问题进行分析，并给出解决方案。

2 问题现象

某客户通过 datax 使用 rdbmsreader 读取 ob 数据并以 orc 格式写入到 hdfs 时，当数据量达到400w 时（act_stock_holder），jkd 使用 2g 堆空间都会导致 OutOfMemoryError，需要显示配置 8G 的堆空间才能同步成功，经验证oracle jdk arm 版 1.8.0_381 和 dragonwell jdk arm 版 1.8.0_332 都是如此。OutOfMemoryError详细的报错堆栈信息如下：

"0-0-0-reader" prio=5 tid=34 RUNNABLE
    at java.lang.OutOfMemoryError.<init>(OutOfMemoryError.java:48)
    at com.alipay.oceanbase.jdbc.MysqlIO.nextRowFast(MysqlIO.java:2657)
       Local Variable: byte[][]#535681
    at com.alipay.oceanbase.jdbc.MysqlIO.nextRow(MysqlIO.java:2320)
    at com.alipay.oceanbase.jdbc.MysqlIO.readSingleRowSet(MysqlIO.java:4207)
       Local Variable: java.util.ArrayList#4
    at com.alipay.oceanbase.jdbc.MysqlIO.getResultSet(MysqlIO.java:572)
       Local Variable: com.alipay.oceanbase.jdbc.Field[]#1
    at com.alipay.oceanbase.jdbc.MysqlIO.readResultsForQueryOrUpdate(MysqlIO.java:3651)
    at com.alipay.oceanbase.jdbc.MysqlIO.readAllResults(MysqlIO.java:2800)
    at com.alipay.oceanbase.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:3220)
       Local Variable: com.alipay.oceanbase.jdbc.MysqlIO#1
       Local Variable: com.alipay.oceanbase.jdbc.Buffer#1
       Local Variable: com.alipay.oceanbase.jdbc.Buffer#2
    at com.alipay.oceanbase.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2903)
       Local Variable: java.lang.String#218
    at com.alipay.oceanbase.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2855)
       Local Variable: java.lang.String#219
    at com.alipay.oceanbase.jdbc.StatementImpl.executeQuery(StatementImpl.java:1445)
       Local Variable: com.alipay.oceanbase.jdbc.StatementImpl$CancelTask#1
    at com.alibaba.datax.plugin.rdbms.util.DBUtil.query(DBUtil.java:471)
    at com.alibaba.datax.plugin.rdbms.util.DBUtil.query(DBUtil.java:443)
       Local Variable: com.alipay.oceanbase.jdbc.StatementImpl#1
    at com.alibaba.datax.plugin.rdbms.util.DBUtil.query(DBUtil.java:422)
    at com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader$Task.startRead(CommonRdbmsReader.java:194)
       Local Variable: com.alipay.oceanbase.jdbc.JDBC4Connection#1
       Local Variable: com.alibaba.datax.common.util.Configuration#1
       Local Variable: java.lang.String#220
       Local Variable: com.alibaba.datax.common.statistics.PerfRecord#1
       Local Variable: com.alibaba.datax.plugin.reader.rdbmsreader.SubCommonRdbmsReader$Task#1
       Local Variable: com.alibaba.datax.core.statistics.plugin.task.StdoutPluginCollector#2
    at com.alibaba.datax.plugin.reader.rdbmsreader.RdbmsReader$Task.startRead(RdbmsReader.java:84)
       Local Variable: com.alibaba.datax.core.transport.exchanger.BufferedRecordExchanger#2
    at com.alibaba.datax.core.taskgroup.runner.ReaderRunner.run(ReaderRunner.java:57)
       Local Variable: com.alibaba.datax.core.taskgroup.runner.ReaderRunner#1
       Local Variable: com.alibaba.datax.plugin.reader.rdbmsreader.RdbmsReader$Task#1
       Local Variable: com.alibaba.datax.common.statistics.PerfRecord#2
       Local Variable: com.alibaba.datax.common.statistics.PerfRecord#4
       Local Variable: com.alibaba.datax.common.statistics.PerfRecord#3
       Local Variable: com.alibaba.datax.common.statistics.PerfRecord#5
    at java.lang.Thread.run(Thread.java:855)

3 问题原因

该问题跟 jdk 版本和 arm 架构无关，而是因为读取 Ob 数据时，默认情况下会一次性读取所有数据并加载到内存后再做后续处理，所以数据量大时会占用大量堆空间甚至oom，目前版本的oceanbasev10reader/rdbmsreader都存在该问题。

4 解决方案

  • 可以在ob的jdbcurl中显示设置参数 useCursorFetch=true (>=5.0版驱动开始支持)，此时底层读取数据时会使用服务器端游标且每次从服务端批量读取 fetch_size 条数据进行处理，从而避免掉大数据量下占用堆空间大甚至 OOM 的问题（该推荐配置与UF30微服务略有不同）.
  • 推荐的OB jdbc url格式如下（读写都可以使用该格式；oceanbasev10writer/oceanbasev10reader/rdbmswriter/rdbmsreader都可以使用该格式）：jdbc:oceanbase://10.20.182.144:2883/sys?rewriteBatchedStatements=true&useCursorFetch=true。
  • 推荐使用 ob 的专用插件 oceanbasev10writer/oceanbasev10reader,作为 ob 专用插件，其底层自动配置了多个参数，比如oceanbasev10writer 会自动配置 rewriteBatchedStatements=ture,比如oceanbasev10reader会自动配置ResultSet.TYPE_FORWARD_ONLY，所以理论上同步性能会更好一些，特别是考虑到后续随着版本升级迭代，还会有一些功能增强问题修复之类的更新，所以推荐优先使用专用插件；

    5 技术背景

  • datax 的 oceanbasev10reader/ oceanbasev10writer 专用插件，在底层会自动设置一些相关参数如 ResultSet.TYPE_FORWARD_ONLY/rewriteBatchedStatements，且当用户没有显示配置readBatchSize会自动设置 DEFAULT_READ_BATCH_SIZE.
  • 相关源码：
    com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderTask#startRead0
com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderTask#doRead
com.alibaba.datax.plugin.writer.oceanbasev10writer.task.SingleTableWriterTask#init
com.alibaba.datax.plugin.writer.oceanbasev10writer.task.SingleTableWriterTask#rewriteSql
com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.OCJConnHolder#initConnection
com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.OBDataSourceV10#buildJdbcProperty

  • 相关参数：rewriteBatchedStatements/useServerPrepStmts/allowMultiQueries/defaultFetchSize/useCursorFetch
  • useCursorFetch: Should the driver use cursor-based fetching to retrieve rows? If set to "true" and 'defaultFetchSize' is set to a value higher than zero or 'setFetchSize()' with a value higher than zero is called on a statement, then the cursor-based result set will be used. Please note that 'useServerPrepStmts' is automatically set to "true" in this case because cursor functionality is available only for server-side prepared statements.
  • 参考链接：
    https://www.oceanbase.com/docs/common-oceanbase-connector-j-cn-10000000001943209
https://dev.mysql.com/doc/connector-j/8.1/en/connector-j-reference-configuration-properties.html
https://www.oceanbase.com/docs/common-oceanbase-connector-j-cn-10000000001944617
https://www.oceanbase.com/docs/enterprise-oceanbase-connector-j-cn-10000000001943207
https://cloud.tencent.com/developer/article/2101083

    6 不同 JDK 的性能差异

  • 使用同样的上游ob数据源和下游hdfs数据源，且同一时间同一台机器分别提交两个datax，一个使用龙井一个使用oracle(oracle jdk arm 版 1.8.0_381 和 dragonwell jdk arm 版 1.8.0_332)，进行对比测试可以发现，在同步大量数据且配置堆空间为8G时，oracleJDK 和龙井 jdk在平均同步速度上差很多，前者只有 330KB/S, 后者能达到2.63MB/S，即龙井jdk arm版比oralce jdk arm更有性能优势；
  • 其底层原因可能跟两者默认的GC参数不同有关(龙井arm默认是ParNew和ConcurrentMarkSweep，而oracle arm默认是PS Scavenge和PS MarkSweep)；
  • 为进一步确认性能差异的原因，可以再指定GC参数对比测试下 oracle/dragonwell 的同步性能，其中 oracle jdk 可以使用如下命令指定GC参数，dragon使用默认GC参数即可（都需要配置JAVA_HOME环境变量）：
    python /opt/DataX/bin/datax.py /tmp/ob.datax  --jvm='-Xms8196m -Xmx8196m -XX:+UseCompressedClassPointers -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -Doracle.jdbc.fanEnabled=false -Duser.timezone=GMT+8'

7. tcpdump- 抓包分析

在分析问题的过程中，我们通过在 datax 节点使用tcpdump 抓包并导入到wireshark进行分析（tcpdump -i any -nn -s 100 "port 2883 or port 8020" -w /tmp/ob.pcap），并发现了如下现象：

  • 在不进行流控限制时（即不配置datax的"speed": { "channel": 5, "byte": 1048576, "record": 10000}），datax 读ob 和 写 hdfs 时都有短时间内打满网络的现象（即TCP ZeroWindow 和 TCP Window Full）；
  • 同时datax 写 hdfs 时会出现有规律地暂停10秒左右的现象，这是因为 orc 是列存储格式，datax 需要累积一批 row data 并计算转换为 orc 的 stripe后（包括stripe底层的index data/stripe footer）才能写入 hdfs。

文章标签：
数据集成 Data Integration
云数据库 RDS MySQL 版
DataX
Java
OceanBase
关系型数据库
Oracle
关键词：
数据集成 Data Integration数据同步
datax数据同步
云数据库OceanBase参数
数据集成 Data Integration参数
数据同步datax
明哥的IT随笔
目录
相关文章
让线程再跑一会
|
6月前
|
SQL 存储 关系型数据库
DataX - 全量数据同步工具（2）
DataX - 全量数据同步工具
让线程再跑一会
629 0
土木林森
|
4月前
|
关系型数据库 MySQL 大数据
DataX：数据同步的超音速英雄！阿里开源工具带你飞越数据传输的银河系，告别等待和故障的恐惧！快来见证这一数据工程的奇迹！
【8月更文挑战第13天】DataX是由阿里巴巴开源的一款专为大规模数据同步设计的工具，在数据工程领域展现强大竞争力。它采用插件化架构，支持多种数据源间的高效迁移。相较于Apache Sqoop和Flume，DataX通过并发写入和流处理实现了高性能同步，并简化了配置流程。DataX还支持故障恢复，能够在同步中断后继续执行，节省时间和资源。这些特性使其成为构建高效可靠数据同步方案的理想选择。
土木林森
380 2
游客e4ooen74o6ems
|
5月前
|
监控 数据挖掘 大数据
阿里云开源利器：DataX3.0——高效稳定的离线数据同步解决方案
对于需要集成多个数据源进行大数据分析的场景，DataX3.0同样提供了有力的支持。企业可以使用DataX将多个数据源的数据集成到一个统一的数据存储系统中，以便进行后续的数据分析和挖掘工作。这种集成能力有助于提升数据分析的效率和准确性，为企业决策提供有力支持。
游客e4ooen74o6ems
1463 0
pbeskoyq7ffm4
|
4月前
|
Java 关系型数据库 DataX
DATAX数据同步
DATAX数据同步
pbeskoyq7ffm4
651 0
嘟嘟嘟嘟嘟嘟
|
5月前
|
DataWorks NoSQL fastjson
DataWorks操作报错合集之DataX进行MongoDB全量迁移的过程中，DataX的MongoDB Reader插件在初始化阶段找不到Fastjson 2.x版本的类库，该怎么办
DataWorks是阿里云提供的一站式大数据开发与治理平台，支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中，可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
嘟嘟嘟嘟嘟嘟
169 5
游客lkw7toijkbadm
|
5月前
|
分布式计算 关系型数据库 MySQL
MySQL超时参数优化与DataX高效数据同步实践
通过合理设置MySQL的超时参数，可以有效地提升数据库的稳定性和性能。而DataX作为一种高效的数据同步工具，可以帮助企业轻松实现不同数据源之间的数据迁移。无论是优化MySQL参数还是使用DataX进行数据同步，都需要根据具体的应用场景来进行细致的配置和测试，以达到最佳效果。
游客lkw7toijkbadm
506 0
冲冲冲c
|
6月前
|
DataWorks Java 调度
DataWorks产品使用合集之进行离线同步时，如何使用DataX的Reader插件来实现源端过滤
DataWorks作为一站式的数据开发与治理平台，提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案，帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述，涵盖数据处理的各个环节。
冲冲冲c
78 0
DataWorks产品使用合集之进行离线同步时，如何使用DataX的Reader插件来实现源端过滤
让线程再跑一会
|
6月前
|
SQL 关系型数据库 MySQL
DataX - 全量数据同步工具（1）
DataX - 全量数据同步工具
让线程再跑一会
946 0
番茄酱脑袋
|
4月前
|
数据采集 DataWorks 监控
DataWorks产品使用合集之数据集成并发数不支持批量修改，该怎么办
DataWorks作为一站式的数据开发与治理平台，提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案，帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述，涵盖数据处理的各个环节。
番茄酱脑袋
56 0
东方睿赢
|
2月前
|
数据采集 DataWorks 数据管理
DataWorks不是Excel，它是一个数据集成和数据管理平台
【10月更文挑战第10天】随着大数据技术的发展，企业对数据处理的需求日益增长。阿里云推出的DataWorks是一款强大的数据集成和管理平台，提供从数据采集、清洗、加工到应用的一站式解决方案。本文通过电商平台案例，详细介绍了DataWorks的核心功能和优势，展示了如何高效处理大规模数据，帮助企业挖掘数据价值。
东方睿赢
134 1

热门文章

最新文章

  • 1
    离线数据同步变迁
  • 2
    OceanBase社区版可以通过Zabbix监控
  • 3
    数据同步原理
  • 4
    厉害了，蚂蚁金服！创造了中国自己的数据库OceanBase
  • 5
    OceanBase 2.0让百万支付不是梦？
  • 6
    OceanBase知识加油站（工具篇二）：使用导数工具了嘛？
  • 7
    OceanBase源码解读（二）：SQL的一生
  • 8
    共建全栈国产化解决方案 | OceanBase持续与合作伙伴推进测试互认
  • 9
    十年磨一剑！支付宝自研数据库OceanBase通过阿里云向全球开放
  • 10
    OceanBase4.0：使用grafana监控oceanbase
  • 1
    离线数据同步变迁
    29
  • 2
    数据同步原理
    87
  • 3
    一文彻底搞定Redis与MySQL的数据同步
    720
  • 4
    实现数据同步的webapi接口
    78
  • 5
    深入了解MySQL主从复制：构建高效稳定的数据同步架构
    143
  • 6
    微信小程序学习之数据绑定，事件绑定，事件传参与数据同步的学习记录
    44
  • 7
    oss数据同步maxcompute报错
    153
  • 8
    震惊！Uno Platform 竟能这样实现实时数据同步，开发者必看的最佳实践秘籍大公开！
    74
  • 9
    【C# 多线程编程陷阱揭秘】：小心！那些让你的程序瞬间崩溃的多线程数据同步异常问题，看完这篇你就能轻松应对！
    195
  • 10
    Canal作为一款高效、可靠的数据同步工具，凭借其基于MySQL binlog的增量同步机制，在数据同步领域展现了强大的应用价值
    841

    • 相关课程

    更多
  • 阿里云 DTS是如何做实时数据集成的？
  • OceanBase社区版入门到实战教程
  • 实时数据接入：5分钟上手 Flink MySQL 连接器

    • 相关电子书

    更多
  • DataWorks数据集成实时同步最佳实践（含内测邀请）-2020飞天大数据平台实战应用第一季
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 《DataWorks 数据集成实时同步》

    • 相关实验场景

    更多
  • 使用DAS实现数据库自动SQL限流
  • PolarDB MySQL自动SQL限流

    • 推荐镜像

    更多
  • oceanbase
  • postgresql
  • clickhouse
    • 下一篇
    DataWorks售前咨询