Datax 支持增量 oracle writeMode update

2024-09-04 00:18
文章标签 oracle 支持 增量 update datax writemode

本文主要是介绍Datax 支持增量 oracle writeMode update，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Datax 支持增量 oracle update

  • datax介绍
    • 支持增量 oracle update
    • 修改 OracleWriter.java
    • 修改WriterUtil.java
    • 修改CommonRdbmsWriter.java
    • 效果
    • 源码

datax介绍

DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。

支持增量 oracle update

我们使用datax 希望支持oracle 增量导入数据：地址：https://gitee.com/cecotw/DataX

链接:https://pan.baidu.com/s/1mbEvLsDZZNWMYrTTTeYkAw 密码:v97c

修改 OracleWriter.java

删除限制
在这里插入图片描述

修改WriterUtil.java

添加oracle 数据插入类型转换：
在这里插入图片描述

    public static String getWriteTemplate(List<String> columnHolders, List<String> valueHolders, String writeMode, DataBaseType dataBaseType, boolean forceUseUpdate) {boolean isWriteModeLegal = writeMode.trim().toLowerCase().startsWith("insert")|| writeMode.trim().toLowerCase().startsWith("replace")|| writeMode.trim().toLowerCase().startsWith("update");if (!isWriteModeLegal) {throw DataXException.asDataXException(DBUtilErrorCode.ILLEGAL_VALUE,String.format("您所配置的 writeMode:%s 错误. 因为DataX 目前仅支持replace,update 或 insert 方式. 请检查您的配置并作出修改.", writeMode));}// && writeMode.trim().toLowerCase().startsWith("replace")String writeDataSqlTemplate;if (forceUseUpdate ||((dataBaseType == DataBaseType.MySql || dataBaseType == DataBaseType.Tddl) && writeMode.trim().toLowerCase().startsWith("update"))) {//update只在mysql下使用writeDataSqlTemplate = new StringBuilder().append("INSERT INTO %s (").append(StringUtils.join(columnHolders, ",")).append(") VALUES(").append(StringUtils.join(valueHolders, ",")).append(")").append(onDuplicateKeyUpdateString(columnHolders)).toString();} else {if (dataBaseType == DataBaseType.Oracle) {writeDataSqlTemplate = new StringBuilder().append(onMergeIntoDoString(writeMode, columnHolders, valueHolders)).append("INSERT (").append(StringUtils.join(columnHolders, ",")).append(") VALUES(").append(StringUtils.join(valueHolders, ",")).append(")").toString();} else {//这里是保护,如果其他错误的使用了update,需要更换为replaceif (writeMode.trim().toLowerCase().startsWith("update")) {writeMode = "replace";}writeDataSqlTemplate = new StringBuilder().append(writeMode).append(" INTO %s (").append(StringUtils.join(columnHolders, ",")).append(") VALUES(").append(StringUtils.join(valueHolders, ",")).append(")").toString();}}return writeDataSqlTemplate;}

增加onMergeIntoDoString方法：
在这里插入图片描述

    public static String onMergeIntoDoString(String merge, List<String> columnHolders, List<String> valueHolders) {String[] sArray = getStrings(merge);StringBuilder sb = new StringBuilder();sb.append("MERGE INTO %s A USING ( SELECT ");boolean first = true;boolean first1 = true;StringBuilder str = new StringBuilder();StringBuilder update = new StringBuilder();for (String columnHolder : columnHolders) {if (Arrays.asList(sArray).contains(columnHolder)) {if (!first) {sb.append(",");str.append(" AND ");} else {first = false;}str.append("TMP.").append(columnHolder);sb.append("?");str.append(" = ");sb.append(" AS ");str.append("A.").append(columnHolder);sb.append(columnHolder);}}for (String columnHolder : columnHolders) {if (!Arrays.asList(sArray).contains(columnHolder)) {if (!first1) {update.append(",");} else {first1 = false;}update.append(columnHolder);update.append(" = ");update.append("?");}}sb.append(" FROM DUAL ) TMP ON (");sb.append(str);sb.append(" ) WHEN MATCHED THEN UPDATE SET ");sb.append(update);sb.append(" WHEN NOT MATCHED THEN ");return sb.toString();}

增加getStrings方法：
在这里插入图片描述

    public static String[] getStrings(String merge) {merge = merge.replace("update", "");merge = merge.replace("(", "");merge = merge.replace(")", "");merge = merge.replace(" ", "");return merge.split(",");}

修改CommonRdbmsWriter.java

在这里插入图片描述

        public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) {this.taskPluginCollector = taskPluginCollector;List<String> columns = new ArrayList<>();List<String> columnsOne = new ArrayList<>();List<String> columnsTwo = new ArrayList<>();if (this.dataBaseType == DataBaseType.Oracle) {String merge = this.writeMode;String[] sArray = WriterUtil.getStrings(merge);int size = this.columns.size();int i = 0;for (int j = 0; j < size; j++) {if (Arrays.asList(sArray).contains(this.columns.get(j))) {columnsOne.add(this.columns.get(j));}}for (int j = 0; j < size; j++) {if (!Arrays.asList(sArray).contains(this.columns.get(j))) {columnsTwo.add(this.columns.get(j));}}for (String column : columnsOne) {columns.add(i, column);i++;}for (String column : columnsTwo) {columns.add(i, column);i++;}}columns.addAll(this.columns);// 用于写入数据的时候的类型根据目的表字段类型转换this.resultSetMetaData = DBUtil.getColumnMetaData(connection,this.table, StringUtils.join(columns, ","));// 写数据库的SQL语句calcWriteRecordSql();List<Record> writeBuffer = new ArrayList<Record>(this.batchSize);int bufferBytes = 0;try {Record record;while ((record = recordReceiver.getFromReader()) != null) {if (record.getColumnNumber() != this.columnNumber && this.dataBaseType != DataBaseType.Oracle) {// 源头读取字段列数与目的表字段写入列数不相等，直接报错throw DataXException.asDataXException(DBUtilErrorCode.CONF_ERROR,String.format("列配置信息有错误. 因为您配置的任务中，源头读取字段数:%s 与 目的表要写入的字段数:%s 不相等. 请检查您的配置并作出修改.",record.getColumnNumber(),this.columnNumber));}writeBuffer.add(record);bufferBytes += record.getMemorySize();if (writeBuffer.size() >= batchSize || bufferBytes >= batchByteSize) {doBatchInsert(connection, writeBuffer);writeBuffer.clear();bufferBytes = 0;}}if (!writeBuffer.isEmpty()) {doBatchInsert(connection, writeBuffer);writeBuffer.clear();bufferBytes = 0;}} catch (Exception e) {throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);} finally {writeBuffer.clear();bufferBytes = 0;DBUtil.closeDBResources(null, null, connection);}}

在这里插入图片描述

        protected void doBatchInsert(Connection connection, List<Record> buffer)throws SQLException {PreparedStatement preparedStatement = null;try {connection.setAutoCommit(false);preparedStatement = connection.prepareStatement(this.writeRecordSql);if (this.dataBaseType == DataBaseType.Oracle) {String merge = this.writeMode;String[] sArray = WriterUtil.getStrings(merge);for (Record record : buffer) {List<Column> recordOne = new ArrayList<>();for (int j = 0; j < this.columns.size(); j++) {if (Arrays.asList(sArray).contains(this.columns.get(j))) {recordOne.add(record.getColumn(j));}}for (int j = 0; j < this.columns.size(); j++) {if (!Arrays.asList(sArray).contains(this.columns.get(j))) {recordOne.add(record.getColumn(j));}}for (int j = 0; j < this.columns.size(); j++) {recordOne.add(record.getColumn(j));}for (int j = 0; j < recordOne.size(); j++) {record.setColumn(j, recordOne.get(j));}preparedStatement = fillPreparedStatement(preparedStatement, record);preparedStatement.addBatch();}} else {for (Record record : buffer) {preparedStatement = fillPreparedStatement(preparedStatement, record);preparedStatement.addBatch();}}preparedStatement.executeBatch();connection.commit();} catch (SQLException e) {LOG.warn("回滚此次写入, 采用每次写入一行方式提交. 因为:" + e.getMessage());connection.rollback();doOneInsert(connection, buffer);} catch (Exception e) {throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);} finally {DBUtil.closeDBResources(preparedStatement, null);}}

效果

{"job": {"setting": {"speed": {"byte": 1048576},"errorLimit": {"record": 0,"percentage": 0.02}},"content": [{"reader": {"name": "postgresqlreader","parameter": {"username": "postgres","password": "postgres","connection": [{"querySql":["SELECT seq,userid,name FROM user"],"jdbcUrl": ["jdbc:postgresql://127.0.0.1:5432/postgres"]}]}},"writer": {"name": "oraclewriter","parameter": {"username": "oracle","password": "oracle","column": ["seq","userid","name"],"connection": [{"jdbcUrl": "jdbc:oracle:thin:@localhost:1521:oracle","table": ["user1"]}],"writeMode": "update (seq,userid)"}}}]}
}

源码

  • 关于 DATAX改造后的代码 ，参考 这儿.（https://gitee.com/cecotw/DataX）

这篇关于Datax 支持增量 oracle writeMode update的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1134451。

相关文章

Oracle type （自定义类型的使用）

Oracle type （自定义类型的使用）

oracle - type   type定义： oracle中自定义数据类型 oracle中有基本的数据类型，如number，varchar2，date，numeric，float....但有时候我们需要特殊的格式， 如将name定义为（firstname,lastname）的形式，我们想把这个作为一个表的一列看待，这时候就要我们自己定义一个数据类型 格式 ：create or repla
阅读更多...
ORACLE 11g 创建数据库时 Enterprise Manager配置失败的解决办法 无法打开OEM的解决办法

ORACLE 11g 创建数据库时 Enterprise Manager配置失败的解决办法 无法打开OEM的解决办法

在win7 64位系统下安装oracle11g，在使用Database configuration Assistant创建数据库时，在创建到85%的时候报错，错误如下： 解决办法： 在listener.ora中增加对BlueAeri-PC或ip地址的侦听，具体步骤如下： 1.启动Net Manager，在“监听程序”--Listener下添加一个地址，主机名写计
阅读更多...
Oracle Start With关键字

Oracle Start With关键字

Oracle Start With关键字 前言 旨在记录一些Oracle使用中遇到的各种各样的问题. 同时希望能帮到和我遇到同样问题的人. Start With (树查询) 问题描述: 在数据库中, 有一种比较常见得 设计模式, 层级结构 设计模式, 具体到 Oracle table中, 字段特点如下: ID, DSC, PID; 三个字段, 分别表示 当前标识的 ID(主键), DSC 当
阅读更多...
oracle分页和mysql分页

oracle分页和mysql分页

mysql 分页 --查前5 数据select * from table_name limit 0,5 select * from table_name limit 5 --limit关键字的用法：LIMIT [offset,] rows--offset指定要返回的第一行的偏移量，rows第二个指定返回行的最大数目。初始行的偏移量是0(不是1)。   oracle 分页 --查前1-9
阅读更多...
Golang支持平滑升级的HTTP服务

Golang支持平滑升级的HTTP服务

前段时间用Golang在做一个HTTP的接口，因编译型语言的特性，修改了代码需要重新编译可执行文件，关闭正在运行的老程序，并启动新程序。对于访问量较大的面向用户的产品，关闭、重启的过程中势必会出现无法访问的情况，从而影响用户体验。 使用Golang的系统包开发HTTP服务，是无法支持平滑升级（优雅重启）的，本文将探讨如何解决该问题。 一、平滑升级（优雅重启）的一般思路 一般情况下，要实现平滑
阅读更多...
ORACLE语法-包(package)、存储过程(procedure)、游标(cursor)以及java对Result结果集的处理

ORACLE语法-包(package)、存储过程(procedure)、游标(cursor)以及java对Result结果集的处理

陈科肇 示例： 包规范 CREATE OR REPLACE PACKAGE PACK_WMS_YX IS-- Author : CKZ-- Created : 2015/8/28 9:52:29-- Purpose : 同步数据-- Public type declarations，游标 退休订单TYPE retCursor IS REF CURSOR;-- RETURN vi_co_co
阅读更多...
Oracle主键和外键详解及实用技巧

Oracle主键和外键详解及实用技巧

在 Oracle 数据库中，主键（Primary Key）和外键（Foreign Key）用于维护数据库表之间的数据完整性。 1. 主键（Primary Key） 主键是一列或多列，能够唯一标识表中的每一行。表中只能有一个主键，并且主键列不能为空（即 NOT NULL）。 特性： 唯一性：主键中的每一个值都是唯一的，不能重复。非空性：主键列不能包含 NULL 值。索引：Oracle 自动为
阅读更多...
sqlite不支持中文排序，采用java排序

sqlite不支持中文排序，采用java排序

方式一 不支持含有重复字段进行排序 /*** sqlite不支持中文排序，改用java排序* 根据指定的对象属性字段，排序对象集合，顺序* @param list* @param field* @return*/public static List sortListByField(List<?> list,String field){List temp = new ArrayList(
阅读更多...
一款支持同一个屏幕界面同时播放多个视频的视频播放软件

一款支持同一个屏幕界面同时播放多个视频的视频播放软件

GridPlayer 是一款基于 VLC 的免费开源跨平台多视频同步播放工具，支持在一块屏幕上同时播放多个视频。其主要功能包括： 多视频播放：用户可以在一个窗口中同时播放任意数量的视频，数量仅受硬件性能限制。支持多种格式和流媒体：GridPlayer 支持所有由 VLC 支持的视频格式以及流媒体 URL（如 m3u8 链接）。自定义网格布局：用户可以配置播放器的网格布局，以适应不同的观看需求。硬
阅读更多...
Science Robotics 首尔国立大学研究团队推出BBEX外骨骼，实现多维力量支持！

Science Robotics 首尔国立大学研究团队推出BBEX外骨骼，实现多维力量支持！

重复性举起物体可能会对脊柱和背部肌肉造成损伤，由此引发的腰椎损伤是工业环境等工作场所中一个普遍且令人关注的问题。为了减轻这类伤害，有研究人员已经研发出在举起任务中为工人提供辅助的背部支撑装置。然而，现有的这类装置通常无法在非对称性的举重过程中提供多维度的力量支持。此外，针对整个人体脊柱的设备安全性验证也一直是一个缺失的环节。 据探索前沿科技边界，传递前沿科技成果的X-robot投稿，来自首尔国立
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2