> getRowKeyElementList() {
+ return rowKeyElementList;
+ }
+}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ObHbaseWriter.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ObHbaseWriter.java
new file mode 100644
index 0000000000..555ce83638
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ObHbaseWriter.java
@@ -0,0 +1,267 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter;
+
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_OB_TABLE_CLIENT_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_OB_TABLE_HBASE_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_COM_ALIPAY_TABLE_CLIENT_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_COM_ALIPAY_TABLE_HBASE_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_HBASE_LOG_PATH;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_TABLE_CLIENT_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_TABLE_CLIENT_PROPERTY;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_TABLE_HBASE_LOG_LEVEL;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.OB_TABLE_HBASE_PROPERTY;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.util.ObVersion;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ext.ServerConnectInfo;
+import com.alibaba.datax.plugin.writer.obhbasewriter.task.ObHBaseWriteTask;
+import com.google.common.base.Preconditions;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+
+/**
+ *
+ */
+public class ObHbaseWriter extends Writer {
+ /**
+ * Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
+ *
+ * 整个 Writer 执行流程是：
+ *
+ *
+ * Job类init-->prepare-->split
+ *
+ * Task类init-->prepare-->startWrite-->post-->destroy
+ * Task类init-->prepare-->startWrite-->post-->destroy
+ *
+ * Job类post-->destroy
+ *
+ */
+ public static class Job extends Writer.Job {
+ private Configuration originalConfig = null;
+ private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：通常在这里对用户的配置进行校验：是否缺失必填项？有无错误值？有没有无关配置项？...
+ * 并给出清晰的报错/警告提示。校验通常建议采用静态工具类进行，以保证本类结构清晰。
+ */
+ @Override
+ public void init() {
+ if (System.getProperty(OB_TABLE_CLIENT_PROPERTY) == null) {
+ LOG.info(OB_TABLE_CLIENT_PROPERTY + " not set");
+ System.setProperty(OB_TABLE_CLIENT_PROPERTY, OB_HBASE_LOG_PATH);
+ }
+ if (System.getProperty(OB_TABLE_HBASE_PROPERTY) == null) {
+ LOG.info(OB_TABLE_HBASE_PROPERTY + " not set");
+ System.setProperty(OB_TABLE_HBASE_PROPERTY, OB_HBASE_LOG_PATH);
+ }
+ if (System.getProperty(OB_TABLE_CLIENT_LOG_LEVEL) == null) {
+ LOG.info(OB_TABLE_CLIENT_LOG_LEVEL + " not set");
+ System.setProperty(OB_TABLE_CLIENT_LOG_LEVEL, DEFAULT_OB_TABLE_CLIENT_LOG_LEVEL);
+ }
+ if (System.getProperty(OB_TABLE_HBASE_LOG_LEVEL) == null) {
+ LOG.info(OB_TABLE_HBASE_LOG_LEVEL + " not set");
+ System.setProperty(OB_TABLE_HBASE_LOG_LEVEL, DEFAULT_OB_TABLE_HBASE_LOG_LEVEL);
+ }
+ if (System.getProperty(OB_COM_ALIPAY_TABLE_CLIENT_LOG_LEVEL) == null) {
+ LOG.info(OB_COM_ALIPAY_TABLE_CLIENT_LOG_LEVEL + " not set");
+ System.setProperty(OB_COM_ALIPAY_TABLE_CLIENT_LOG_LEVEL, DEFAULT_OB_TABLE_CLIENT_LOG_LEVEL);
+ }
+ if (System.getProperty(OB_COM_ALIPAY_TABLE_HBASE_LOG_LEVEL) == null) {
+ LOG.info(OB_COM_ALIPAY_TABLE_HBASE_LOG_LEVEL + " not set");
+ System.setProperty(OB_COM_ALIPAY_TABLE_HBASE_LOG_LEVEL, DEFAULT_OB_TABLE_HBASE_LOG_LEVEL);
+ }
+
+ LOG.info("{} is set to {}, {} is set to {}",
+ OB_TABLE_CLIENT_PROPERTY, OB_HBASE_LOG_PATH, OB_TABLE_HBASE_PROPERTY, OB_HBASE_LOG_PATH);
+ this.originalConfig = super.getPluginJobConf();
+ boolean useOdpMode = originalConfig.getBool(ConfigKey.USE_ODP_MODE, false);
+ String configUrl = originalConfig.getString(ConfigKey.OBCONFIG_URL, null);
+ String jdbcUrl = originalConfig.getString(ConfigKey.JDBC_URL, null);
+ jdbcUrl = DataBaseType.MySql.appendJDBCSuffixForReader(jdbcUrl);
+ String user = originalConfig.getString(Key.USERNAME, null);
+ String password = originalConfig.getString(Key.PASSWORD);
+ ServerConnectInfo serverConnectInfo = new ServerConnectInfo(jdbcUrl, user, password);
+ if (useOdpMode) {
+ originalConfig.set(ConfigKey.ODP_HOST, serverConnectInfo.host);
+ originalConfig.set(ConfigKey.ODP_PORT, serverConnectInfo.port);
+ } else if (StringUtils.isBlank(configUrl)) {
+ serverConnectInfo.setSysUser(originalConfig.getString(ConfigKey.OB_SYS_USER));
+ serverConnectInfo.setSysPass(originalConfig.getString(ConfigKey.OB_SYS_PASSWORD));
+ try {
+ originalConfig.set(ConfigKey.OBCONFIG_URL, queryRsUrl(serverConnectInfo));
+ originalConfig.set(ConfigKey.OB_SYS_USER, serverConnectInfo.sysUser);
+ originalConfig.set(ConfigKey.OB_SYS_PASSWORD, serverConnectInfo.sysPass);
+ LOG.info("fetch configUrl success, configUrl is {}", configUrl);
+ } catch (Exception e) {
+ LOG.error("fail to get configure url: " + e.getMessage());
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.REQUIRED_VALUE, "Missing obConfigUrl");
+ }
+ }
+ if (StringUtils.isBlank(originalConfig.getString(ConfigKey.DBNAME))) {
+ originalConfig.set(ConfigKey.DBNAME, serverConnectInfo.databaseName);
+ }
+ ConfigValidator.validateParameter(this.originalConfig);
+ }
+
+ private String queryRsUrl(ServerConnectInfo serverInfo) {
+ String configUrl = originalConfig.getString(ConfigKey.OBCONFIG_URL, null);
+ if (configUrl == null) {
+ try {
+ Connection conn = null;
+ int retry = 0;
+ final String sysJDBCUrl = serverInfo.jdbcUrl.replace(serverInfo.databaseName, "oceanbase");
+ do {
+ try {
+ if (retry > 0) {
+ int sleep = retry > 9 ? 500 : 1 << retry;
+ try {
+ TimeUnit.SECONDS.sleep(sleep);
+ } catch (InterruptedException e) {
+ }
+ LOG.warn("retry fetch RsUrl the {} times", retry);
+ }
+ conn = DBUtil.getConnection(DataBaseType.OceanBase, sysJDBCUrl, serverInfo.sysUser, serverInfo.sysPass);
+ String sql = "show parameters like 'obconfig_url'";
+ LOG.info("query param: {}", sql);
+ PreparedStatement stmt = conn.prepareStatement(sql);
+ ResultSet result = stmt.executeQuery();
+ if (result.next()) {
+ configUrl = result.getString("Value");
+ }
+ if (StringUtils.isNotBlank(configUrl)) {
+ break;
+ }
+ } catch (Exception e) {
+ ++retry;
+ LOG.warn("fetch root server list(rsList) error {}", e.getMessage());
+ } finally {
+ DBUtil.closeDBResources(null, conn);
+ }
+ } while (retry < 3);
+
+ LOG.info("configure url is: " + configUrl);
+ originalConfig.set(ConfigKey.OBCONFIG_URL, configUrl);
+ } catch (Exception e) {
+ LOG.error("Fail to get configure url: {}", e.getMessage(), e);
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.REQUIRED_VALUE, "未配置obConfigUrl，且无法获取obConfigUrl");
+ }
+ }
+ return configUrl;
+ }
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+ */
+ // 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
+ @Override
+ public void prepare() {
+ }
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：通常采用工具静态类完成把 Job 配置切分成多个 Task 配置的工作。 这里的
+ * mandatoryNumber 是强制必须切分的份数。
+ */
+ @Override
+ public List split(int mandatoryNumber) {
+ // This function does not need any change.
+ Configuration simplifiedConf = this.originalConfig;
+
+ List splitResultConfigs = new ArrayList();
+ for (int j = 0; j < mandatoryNumber; j++) {
+ splitResultConfigs.add(simplifiedConf.clone());
+ }
+ return splitResultConfigs;
+ }
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之后的后续处理，可以在此处完成。
+ */
+ @Override
+ public void post() {
+ // No post supported
+ }
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：通常配合 Job 中的 post() 方法一起完成 Job 的资源释放。
+ */
+ @Override
+ public void destroy() {
+
+ }
+ }
+
+ public static class Task extends Writer.Task {
+ private Configuration taskConfig;
+ private CommonRdbmsWriter.Task writerTask;
+
+ /**
+ * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处通过对 taskConfig 配置的读取，进而初始化一些资源为
+ * startWrite()做准备。
+ */
+ @Override
+ public void init() {
+ this.taskConfig = super.getPluginJobConf();
+ String mode = this.taskConfig.getString(ConfigKey.MODE);
+ ModeType modeType = ModeType.getByTypeName(mode);
+
+ switch (modeType) {
+ case Normal:
+ try {
+ this.writerTask = new ObHBaseWriteTask(this.taskConfig);
+ } catch (Exception e) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.INIT_ERROR, "ObHbase writer init error:" + e.getMessage());
+ }
+ break;
+ default:
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.ILLEGAL_VALUE, "ObHbase not support this mode type:" + modeType);
+ }
+ }
+
+ /**
+ * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task
+ * 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+ */
+ @Override
+ public void prepare() {
+ this.writerTask.prepare(taskConfig);
+ }
+
+ /**
+ * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。
+ */
+ public void startWrite(RecordReceiver recordReceiver) {
+ this.writerTask.startWrite(recordReceiver, taskConfig, super.getTaskPluginCollector());
+ }
+
+ /**
+ * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task 中有需要进行数据同步之后的后续处理，可以在此处完成。
+ */
+ @Override
+ public void post() {
+ this.writerTask.post(taskConfig);
+ }
+
+ /**
+ * 注意：此方法每个 Task 都会执行一次。 最佳实践：通常配合Task 中的 post() 方法一起完成 Task 的资源释放。
+ */
+ @Override
+ public void destroy() {
+ this.writerTask.destroy(taskConfig);
+ }
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings.properties
new file mode 100644
index 0000000000..63a53efab7
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684table\u4e0d\u5b58\u5728, \u7b97\u51fa\u7684tableName={0},db={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_en_US.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_en_US.properties
new file mode 100644
index 0000000000..86b6a8b3b8
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_en_US.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=The [table] calculated based on the rules does not exist. The calculated [tableName]={0}, [db]={1}. Please check the rules you configured.
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_ja_JP.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_ja_JP.properties
new file mode 100644
index 0000000000..63a53efab7
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_ja_JP.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684table\u4e0d\u5b58\u5728, \u7b97\u51fa\u7684tableName={0},db={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_CN.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_CN.properties
new file mode 100644
index 0000000000..63a53efab7
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_CN.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684table\u4e0d\u5b58\u5728, \u7b97\u51fa\u7684tableName={0},db={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_HK.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_HK.properties
new file mode 100644
index 0000000000..4940a177b6
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_HK.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684table\u4e0d\u5b58\u5728, \u7b97\u51fa\u7684tableName={0},db={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.databasewriterbuffer.1=通過規則計算出來的table不存在, 算出的tableName={0},db={1}, 請檢查您配置的規則.
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_TW.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_TW.properties
new file mode 100644
index 0000000000..4940a177b6
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/LocalStrings_zh_TW.properties
@@ -0,0 +1 @@
+databasewriterbuffer.1=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684table\u4e0d\u5b58\u5728, \u7b97\u51fa\u7684tableName={0},db={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.databasewriterbuffer.1=通過規則計算出來的table不存在, 算出的tableName={0},db={1}, 請檢查您配置的規則.
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObDataSourceErrorCode.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObDataSourceErrorCode.java
new file mode 100644
index 0000000000..5e656a6e41
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObDataSourceErrorCode.java
@@ -0,0 +1,30 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.ext;
+
+import com.alibaba.datax.common.spi.ErrorCode;
+
+public enum ObDataSourceErrorCode implements ErrorCode {
+ DESC("ObDataSourceError code", "connect error");
+
+ private final String code;
+ private final String describe;
+
+ private ObDataSourceErrorCode(String code, String describe) {
+ this.code = code;
+ this.describe = describe;
+ }
+
+ @Override
+ public String getCode() {
+ return this.code;
+ }
+
+ @Override
+ public String getDescription() {
+ return this.describe;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("Code:[%s], Describe:[%s]. ", this.code, this.describe);
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObHbaseTableHolder.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObHbaseTableHolder.java
new file mode 100644
index 0000000000..7b74b9479c
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ObHbaseTableHolder.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2021 OceanBase ob-loader-dumper is licensed under Mulan PSL v2. You can use this software according to
+ * the terms and conditions of the Mulan PSL v2. You may obtain a copy of Mulan PSL v2 at:
+ *
+ * http://license.coscl.org.cn/MulanPSL2
+ *
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING
+ * BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. See the Mulan PSL v2 for more
+ * details.
+ */
+package com.alibaba.datax.plugin.writer.obhbasewriter.ext;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Hbase094xWriterErrorCode;
+import com.alipay.oceanbase.hbase.OHTable;
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author cjyyz
+ * @date 2023/03/16
+ * @since
+ */
+public class ObHbaseTableHolder {
+ private static final Logger LOG = LoggerFactory.getLogger(ObHbaseTableHolder.class);
+
+ private Configuration configuration;
+
+ private String hbaseTableName;
+
+ private OHTable ohTable;
+
+ public ObHbaseTableHolder(Configuration configuration, String hbaseTableName) {
+ this.configuration = configuration;
+ this.hbaseTableName = hbaseTableName;
+ }
+
+ public OHTable getOhTable() {
+ try {
+ if (ohTable == null) {
+ ohTable = new OHTable(configuration, hbaseTableName);
+ }
+ return ohTable;
+ } catch (Exception e) {
+ LOG.error("build obHTable: {} failed. reason: {}", hbaseTableName, e.getMessage());
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.GET_HBASE_TABLE_ERROR, Hbase094xWriterErrorCode.GET_HBASE_TABLE_ERROR.getDescription());
+ }
+ }
+
+ public void destroy() {
+ try {
+ if (ohTable != null) {
+ ohTable.close();
+ }
+ } catch (Exception e) {
+ LOG.warn("error in closing htable: {}. Reason: {}", hbaseTableName, e.getMessage());
+ }
+ }
+}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ServerConnectInfo.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ServerConnectInfo.java
new file mode 100644
index 0000000000..80c2d0d9f7
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/ext/ServerConnectInfo.java
@@ -0,0 +1,146 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.ext;
+
+import com.google.common.base.Preconditions;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+
+public class ServerConnectInfo {
+
+ public String clusterName;
+ public String tenantName;
+ // userName doesn't contain tenantName or clusterName
+ public String userName;
+ public String password;
+ public String databaseName;
+ public String ipPort;
+ public String jdbcUrl;
+ public String host;
+ public String port;
+ public boolean publicCloud;
+ public int rpcPort;
+ public String sysUser;
+ public String sysPass;
+
+ /**
+ *
+ * @param jdbcUrl format is jdbc:oceanbase//ip:port
+ * @param username format is cluster:tenant:username or username@tenant#cluster or user@tenant or user
+ * @param password
+ */
+ public ServerConnectInfo(final String jdbcUrl, final String username, final String password) {
+ this(jdbcUrl, username, password, null, null);
+ }
+
+ public ServerConnectInfo(final String jdbcUrl, final String username, final String password, final String sysUser, final String sysPass) {
+ if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+ String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+ Preconditions.checkArgument(ss.length == 3, "jdbc url format is not correct:" + jdbcUrl);
+ this.userName = username;
+ this.clusterName = ss[1].trim().split(":")[0];
+ this.tenantName = ss[1].trim().split(":")[1];
+ this.jdbcUrl = ss[2];
+ } else {
+ this.jdbcUrl = jdbcUrl;
+ }
+ this.password = password;
+ this.sysUser = sysUser;
+ this.sysPass = sysPass;
+ parseJdbcUrl(jdbcUrl);
+ parseFullUserName(username);
+ }
+
+ private void parseJdbcUrl(final String jdbcUrl) {
+ Pattern pattern = Pattern.compile("//([\\w\\.\\-]+:\\d+)/([\\w-]+)\\?");
+ Matcher matcher = pattern.matcher(jdbcUrl);
+ if (matcher.find()) {
+ String ipPort = matcher.group(1);
+ String dbName = matcher.group(2);
+ this.ipPort = ipPort;
+ String[] hostPort = ipPort.split(":");
+ this.host = hostPort[0];
+ this.port = hostPort[1];
+ this.databaseName = dbName;
+ this.publicCloud = host.endsWith("aliyuncs.com");
+ } else {
+ throw new RuntimeException("Invalid argument:" + jdbcUrl);
+ }
+ }
+
+ private void parseFullUserName(final String fullUserName) {
+ int tenantIndex = fullUserName.indexOf("@");
+ int clusterIndex = fullUserName.indexOf("#");
+ // 适用于jdbcUrl以||_dsc_ob10_dsc_开头的场景
+ if (fullUserName.contains(":") && tenantIndex < 0) {
+ String[] names = fullUserName.split(":");
+ if (names.length != 3) {
+ throw new RuntimeException("invalid argument: " + fullUserName);
+ } else {
+ this.clusterName = names[0];
+ this.tenantName = names[1];
+ this.userName = names[2];
+ }
+ } else if (tenantIndex < 0) {
+ // 适用于short jdbcUrl，且username中不含租户名（主要是公有云场景，此场景下不计算分区）
+ this.userName = fullUserName;
+ this.clusterName = EMPTY;
+ this.tenantName = EMPTY;
+ } else {
+ // 适用于short jdbcUrl，且username中含租户名
+ this.userName = fullUserName.substring(0, tenantIndex);
+ if (clusterIndex < 0) {
+ this.clusterName = EMPTY;
+ this.tenantName = fullUserName.substring(tenantIndex + 1);
+ } else {
+ this.clusterName = fullUserName.substring(clusterIndex + 1);
+ this.tenantName = fullUserName.substring(tenantIndex + 1, clusterIndex);
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "ServerConnectInfo{" +
+ "clusterName='" + clusterName + '\'' +
+ ", tenantName='" + tenantName + '\'' +
+ ", userName='" + userName + '\'' +
+ ", password='" + password + '\'' +
+ ", databaseName='" + databaseName + '\'' +
+ ", ipPort='" + ipPort + '\'' +
+ ", jdbcUrl='" + jdbcUrl + '\'' +
+ ", publicCloud=" + publicCloud +
+ ", rpcPort=" + rpcPort +
+ '}';
+ }
+
+ public String getFullUserName() {
+ StringBuilder builder = new StringBuilder();
+ builder.append(userName);
+ if (publicCloud || (rpcPort != 0 && EMPTY.equals(clusterName))) {
+ return builder.toString();
+ }
+ if (!EMPTY.equals(tenantName)) {
+ builder.append("@").append(tenantName);
+ }
+
+ if (!EMPTY.equals(clusterName)) {
+ builder.append("#").append(clusterName);
+ }
+ if (EMPTY.equals(this.clusterName) && EMPTY.equals(this.tenantName)) {
+ return this.userName;
+ }
+ return builder.toString();
+ }
+
+ public void setRpcPort(int rpcPort) {
+ this.rpcPort = rpcPort;
+ }
+
+ public void setSysUser(String sysUser) {
+ this.sysUser = sysUser;
+ }
+
+ public void setSysPass(String sysPass) {
+ this.sysPass = sysPass;
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings.properties
new file mode 100644
index 0000000000..d41f6151e9
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings.properties
@@ -0,0 +1,21 @@
+multitablewritertask.1=\u914d\u7f6e\u7684tableList\u4e3a\u591a\u8868\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.2=\u914d\u7f6e\u7684\u591a\u5e93\u4e2d\u7684\u8868\u540d\u6709\u91cd\u590d\u7684\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\u548c\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.3=\u914d\u7f6e\u7684\u6240\u6709\u8868\u540d\u90fd\u76f8\u540c\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.4=\u914d\u7f6e\u7684table\u548cdb\u540d\u79f0\u90fd\u76f8\u540c\uff0c\u6b64\u79cd\u56de\u6d41\u65b9\u5f0f\u4e0d\u652f\u6301
+multitablewritertask.5=\u5217\u914d\u7f6e\u4fe1\u606f\u6709\u9519\u8bef. \u56e0\u4e3a\u60a8\u914d\u7f6e\u7684\u4efb\u52a1\u4e2d\uff0c\u6e90\u5934\u8bfb\u53d6\u5b57\u6bb5\u6570:{0} \u4e0e \u76ee\u7684\u8868\u8981\u5199\u5165\u7684\u5b57\u6bb5\u6570:{1} \u4e0d\u76f8\u7b49. \u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e\u5e76\u4f5c\u51fa\u4fee\u6539.
+multitablewritertask.6=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684tableName\u67e5\u627e\u5bf9\u5e94\u7684db\u4e0d\u5b58\u5728\uff0ctableName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.7=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u548ctable\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0},tableName={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.8=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.9=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684dbName[{0}], \u5b58\u5728\u591a\u5f20\u5206\u8868\uff0c\u8bf7\u914d\u7f6e\u60a8\u7684\u5206\u8868\u89c4\u5219.
+multitablewritertask.10=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0},ErrorCode:{1}
+multitablewritertask.11=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0},ErrorCode:{1}
+multitablewritertask.12=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0},ErrorCode:{1}
+multitablewritertask.13=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0},ErrorCode:{1}
+multitablewritertask.14=\u5199\u5165\u8868[{0}]\u5931\u8d25,\u4f11\u7720[{1}]\u6beb\u79d2,\u6570\u636e:{2}
+multitablewritertask.15=\u5199\u5165\u8868[{0}]\u5b58\u5728\u810f\u6570\u636e,record={1}, \u5199\u5165\u5f02\u5e38\u4e3a:
+
+
+singletablewritertask.1=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0},ErrorCode:{1}
+singletablewritertask.2=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0},ErrorCode:{1}
+singletablewritertask.3=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0},ErrorCode:{1}
+singletablewritertask.4=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0},ErrorCode:{1}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_en_US.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_en_US.properties
new file mode 100644
index 0000000000..010db531ac
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_en_US.properties
@@ -0,0 +1,21 @@
+multitablewritertask.1=The configured [tableList] contains multiple tables but no table splitting rules have been configured. Please check your configuration.
+multitablewritertask.2=There are repeated table names in the multiple tables you configured, but no database or table splitting rules have been configured. Please check your configuration.
+multitablewritertask.3=All configured tables share the same name, but no database splitting rules have been configured. Please check your configuration.
+multitablewritertask.4=The configured table and database share the same name. This back-to-source method is not supported.
+multitablewritertask.5=Error in column configuration information. In your configured tasks, the number of source fields to be read: {0} and the number of fields to be written to the target table: {1} are not equivalent. Please check your configuration and make corrections.
+multitablewritertask.6=The database that corresponds to the [tableName] calculated based on the rules does not exist. The [tableName]={0}. Please check the rules you configured.
+multitablewritertask.7=The database and [table] calculated based on the rules do not exist. The calculated [dbName]={0}, and [tableName]={1}. Please check the rules you configured.
+multitablewritertask.8=The database calculated based on the rules does not exist. The calculated [dbName]={0}. Please check the rules you configured.
+multitablewritertask.9=The [dbName] [{0}] calculated based on the rules contains multiple sub-tables. Please configure your table splitting rules.
+multitablewritertask.10=Fatal exception in OB. Roll back this write and hibernate for five minutes. SQLState: {0}. ErrorCode: {1}
+multitablewritertask.11=Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: {0}. ErrorCode: {1}
+multitablewritertask.12=Exception in OB. Roll back this write and hibernate for one second. Write and submit the records one by one. SQLState: {0}. ErrorCode: {1}
+multitablewritertask.13=Exception in OB. Roll back this write. Write and submit the records one by one. SQLState: {0}. ErrorCode: {1}
+multitablewritertask.14=Failed to write to table: [{0}]. Hibernate for [{1}] milliseconds. Data: {2}
+multitablewritertask.15=writing table [{0}] contains dirty data. Record={1}. Writing exception is:
+
+
+singletablewritertask.1=Fatal exception in OB. Roll back this write and hibernate for five minutes. SQLState: {0}. ErrorCode: {1}
+singletablewritertask.2=Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: {0}. ErrorCode: {1}
+singletablewritertask.3=Exception in OB. Roll back this write and hibernate for one second. Write and submit the records one by one. SQLState: {0}. ErrorCode: {1}
+singletablewritertask.4=Exception in OB. Roll back this write. Write and submit the records one by one. SQLState: {0}. ErrorCode: {1}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_ja_JP.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_ja_JP.properties
new file mode 100644
index 0000000000..12e3e481f8
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_ja_JP.properties
@@ -0,0 +1,21 @@
+multitablewritertask.1=\u914d\u7f6e\u7684tableList\u4e3a\u591a\u8868\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.2=\u914d\u7f6e\u7684\u591a\u5e93\u4e2d\u7684\u8868\u540d\u6709\u91cd\u590d\u7684\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\u548c\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.3=\u914d\u7f6e\u7684\u6240\u6709\u8868\u540d\u90fd\u76f8\u540c\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.4=\u914d\u7f6e\u7684table\u548cdb\u540d\u79f0\u90fd\u76f8\u540c\uff0c\u6b64\u79cd\u56de\u6d41\u65b9\u5f0f\u4e0d\u652f\u6301
+multitablewritertask.5=\u5217\u914d\u7f6e\u4fe1\u606f\u6709\u9519\u8bef. \u56e0\u4e3a\u60a8\u914d\u7f6e\u7684\u4efb\u52a1\u4e2d\uff0c\u6e90\u5934\u8bfb\u53d6\u5b57\u6bb5\u6570:{0} \u4e0e \u76ee\u7684\u8868\u8981\u5199\u5165\u7684\u5b57\u6bb5\u6570:{1} \u4e0d\u76f8\u7b49. \u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e\u5e76\u4f5c\u51fa\u4fee\u6539.
+multitablewritertask.6=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684tableName\u67e5\u627e\u5bf9\u5e94\u7684db\u4e0d\u5b58\u5728\uff0ctableName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.7=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u548ctable\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0},tableName={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.8=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.9=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684dbName[{0}], \u5b58\u5728\u591a\u5f20\u5206\u8868\uff0c\u8bf7\u914d\u7f6e\u60a8\u7684\u5206\u8868\u89c4\u5219.
+multitablewritertask.10=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+multitablewritertask.11=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+multitablewritertask.12=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.13=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.14=\u5199\u5165\u8868[{0}]\u5931\u8d25,\u4f11\u7720[{1}]\u6beb\u79d2,\u6570\u636e:{2}
+multitablewritertask.15=\u5199\u5165\u8868[{0}]\u5b58\u5728\u810f\u6570\u636e,record={1}, \u5199\u5165\u5f02\u5e38\u4e3a:
+
+
+singletablewritertask.1=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+singletablewritertask.2=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+singletablewritertask.3=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+singletablewritertask.4=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_CN.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_CN.properties
new file mode 100644
index 0000000000..12e3e481f8
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_CN.properties
@@ -0,0 +1,21 @@
+multitablewritertask.1=\u914d\u7f6e\u7684tableList\u4e3a\u591a\u8868\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.2=\u914d\u7f6e\u7684\u591a\u5e93\u4e2d\u7684\u8868\u540d\u6709\u91cd\u590d\u7684\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\u548c\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.3=\u914d\u7f6e\u7684\u6240\u6709\u8868\u540d\u90fd\u76f8\u540c\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.4=\u914d\u7f6e\u7684table\u548cdb\u540d\u79f0\u90fd\u76f8\u540c\uff0c\u6b64\u79cd\u56de\u6d41\u65b9\u5f0f\u4e0d\u652f\u6301
+multitablewritertask.5=\u5217\u914d\u7f6e\u4fe1\u606f\u6709\u9519\u8bef. \u56e0\u4e3a\u60a8\u914d\u7f6e\u7684\u4efb\u52a1\u4e2d\uff0c\u6e90\u5934\u8bfb\u53d6\u5b57\u6bb5\u6570:{0} \u4e0e \u76ee\u7684\u8868\u8981\u5199\u5165\u7684\u5b57\u6bb5\u6570:{1} \u4e0d\u76f8\u7b49. \u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e\u5e76\u4f5c\u51fa\u4fee\u6539.
+multitablewritertask.6=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684tableName\u67e5\u627e\u5bf9\u5e94\u7684db\u4e0d\u5b58\u5728\uff0ctableName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.7=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u548ctable\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0},tableName={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.8=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.9=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684dbName[{0}], \u5b58\u5728\u591a\u5f20\u5206\u8868\uff0c\u8bf7\u914d\u7f6e\u60a8\u7684\u5206\u8868\u89c4\u5219.
+multitablewritertask.10=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+multitablewritertask.11=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+multitablewritertask.12=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.13=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.14=\u5199\u5165\u8868[{0}]\u5931\u8d25,\u4f11\u7720[{1}]\u6beb\u79d2,\u6570\u636e:{2}
+multitablewritertask.15=\u5199\u5165\u8868[{0}]\u5b58\u5728\u810f\u6570\u636e,record={1}, \u5199\u5165\u5f02\u5e38\u4e3a:
+
+
+singletablewritertask.1=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+singletablewritertask.2=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+singletablewritertask.3=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+singletablewritertask.4=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_HK.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_HK.properties
new file mode 100644
index 0000000000..4c0f8e55bc
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_HK.properties
@@ -0,0 +1,41 @@
+multitablewritertask.1=\u914d\u7f6e\u7684tableList\u4e3a\u591a\u8868\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.2=\u914d\u7f6e\u7684\u591a\u5e93\u4e2d\u7684\u8868\u540d\u6709\u91cd\u590d\u7684\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\u548c\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.3=\u914d\u7f6e\u7684\u6240\u6709\u8868\u540d\u90fd\u76f8\u540c\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.4=\u914d\u7f6e\u7684table\u548cdb\u540d\u79f0\u90fd\u76f8\u540c\uff0c\u6b64\u79cd\u56de\u6d41\u65b9\u5f0f\u4e0d\u652f\u6301
+multitablewritertask.5=\u5217\u914d\u7f6e\u4fe1\u606f\u6709\u9519\u8bef. \u56e0\u4e3a\u60a8\u914d\u7f6e\u7684\u4efb\u52a1\u4e2d\uff0c\u6e90\u5934\u8bfb\u53d6\u5b57\u6bb5\u6570:{0} \u4e0e \u76ee\u7684\u8868\u8981\u5199\u5165\u7684\u5b57\u6bb5\u6570:{1} \u4e0d\u76f8\u7b49. \u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e\u5e76\u4f5c\u51fa\u4fee\u6539.
+multitablewritertask.6=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684tableName\u67e5\u627e\u5bf9\u5e94\u7684db\u4e0d\u5b58\u5728\uff0ctableName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.7=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u548ctable\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0},tableName={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.8=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.9=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684dbName[{0}], \u5b58\u5728\u591a\u5f20\u5206\u8868\uff0c\u8bf7\u914d\u7f6e\u60a8\u7684\u5206\u8868\u89c4\u5219.
+multitablewritertask.10=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+multitablewritertask.11=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+multitablewritertask.12=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.13=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.14=\u5199\u5165\u8868[{0}]\u5931\u8d25,\u4f11\u7720[{1}]\u6beb\u79d2,\u6570\u636e:{2}
+multitablewritertask.15=\u5199\u5165\u8868[{0}]\u5b58\u5728\u810f\u6570\u636e,record={1}, \u5199\u5165\u5f02\u5e38\u4e3a:
+
+
+singletablewritertask.1=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+singletablewritertask.2=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+singletablewritertask.3=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+singletablewritertask.4=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}multitablewritertask.1=配置的tableList為多表，但未配置分表規則，請檢查您的配置
+multitablewritertask.2=配置的多庫中的表名有重複的，但未配置分庫規則和分表規則，請檢查您的配置
+multitablewritertask.3=配置的所有表名都相同，但未配置分庫規則，請檢查您的配置
+multitablewritertask.4=配置的table和db名稱都相同，此種回流方式不支援
+multitablewritertask.5=列配置資訊有錯誤. 因為您配置的任務中，源頭讀取欄位數:{0}與 目的表要寫入的欄位數:{1}不相等. 請檢查您的配置並作出修改.
+multitablewritertask.6=通過規則計算出來的tableName查找對應的db不存在，tableName={0}, 請檢查您配置的規則.
+multitablewritertask.7=通過規則計算出來的db和table不存在，算出的dbName={0},tableName={1}, 請檢查您配置的規則.
+multitablewritertask.8=通過規則計算出來的db不存在，算出的dbName={0}, 請檢查您配置的規則.
+multitablewritertask.9=通過規則計算出來的dbName[{0}], 存在多張分表，請配置您的分表規則.
+multitablewritertask.10=遇到OB致命異常,回滾此次寫入, 休眠 5分鐘,SQLState:{0}
+multitablewritertask.11=遇到OB可恢復異常,回滾此次寫入, 休眠 1分鐘,SQLState:{0}
+multitablewritertask.12=遇到OB異常,回滾此次寫入, 休眠 1秒,採用逐條寫入提交,SQLState:{0}
+multitablewritertask.13=遇到OB異常,回滾此次寫入, 採用逐條寫入提交,SQLState:{0}
+multitablewritertask.14=寫入表[{0}]失敗,休眠[{1}]毫秒,數據:{2}
+multitablewritertask.15=寫入表[{0}]存在髒數據,record={1}, 寫入異常為:
+
+
+singletablewritertask.1=遇到OB致命異常,回滾此次寫入, 休眠 5分鐘,SQLState:{0}
+singletablewritertask.2=遇到OB可恢復異常,回滾此次寫入, 休眠 1分鐘,SQLState:{0}
+singletablewritertask.3=遇到OB異常,回滾此次寫入, 休眠 1秒,採用逐條寫入提交,SQLState:{0}
+singletablewritertask.4=遇到OB異常,回滾此次寫入, 採用逐條寫入提交,SQLState:{0}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_TW.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_TW.properties
new file mode 100644
index 0000000000..4c0f8e55bc
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/LocalStrings_zh_TW.properties
@@ -0,0 +1,41 @@
+multitablewritertask.1=\u914d\u7f6e\u7684tableList\u4e3a\u591a\u8868\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.2=\u914d\u7f6e\u7684\u591a\u5e93\u4e2d\u7684\u8868\u540d\u6709\u91cd\u590d\u7684\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\u548c\u5206\u8868\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.3=\u914d\u7f6e\u7684\u6240\u6709\u8868\u540d\u90fd\u76f8\u540c\uff0c\u4f46\u672a\u914d\u7f6e\u5206\u5e93\u89c4\u5219\uff0c\u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e
+multitablewritertask.4=\u914d\u7f6e\u7684table\u548cdb\u540d\u79f0\u90fd\u76f8\u540c\uff0c\u6b64\u79cd\u56de\u6d41\u65b9\u5f0f\u4e0d\u652f\u6301
+multitablewritertask.5=\u5217\u914d\u7f6e\u4fe1\u606f\u6709\u9519\u8bef. \u56e0\u4e3a\u60a8\u914d\u7f6e\u7684\u4efb\u52a1\u4e2d\uff0c\u6e90\u5934\u8bfb\u53d6\u5b57\u6bb5\u6570:{0} \u4e0e \u76ee\u7684\u8868\u8981\u5199\u5165\u7684\u5b57\u6bb5\u6570:{1} \u4e0d\u76f8\u7b49. \u8bf7\u68c0\u67e5\u60a8\u7684\u914d\u7f6e\u5e76\u4f5c\u51fa\u4fee\u6539.
+multitablewritertask.6=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684tableName\u67e5\u627e\u5bf9\u5e94\u7684db\u4e0d\u5b58\u5728\uff0ctableName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.7=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u548ctable\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0},tableName={1}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.8=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684db\u4e0d\u5b58\u5728\uff0c\u7b97\u51fa\u7684dbName={0}, \u8bf7\u68c0\u67e5\u60a8\u914d\u7f6e\u7684\u89c4\u5219.
+multitablewritertask.9=\u901a\u8fc7\u89c4\u5219\u8ba1\u7b97\u51fa\u6765\u7684dbName[{0}], \u5b58\u5728\u591a\u5f20\u5206\u8868\uff0c\u8bf7\u914d\u7f6e\u60a8\u7684\u5206\u8868\u89c4\u5219.
+multitablewritertask.10=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+multitablewritertask.11=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+multitablewritertask.12=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.13=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+multitablewritertask.14=\u5199\u5165\u8868[{0}]\u5931\u8d25,\u4f11\u7720[{1}]\u6beb\u79d2,\u6570\u636e:{2}
+multitablewritertask.15=\u5199\u5165\u8868[{0}]\u5b58\u5728\u810f\u6570\u636e,record={1}, \u5199\u5165\u5f02\u5e38\u4e3a:
+
+
+singletablewritertask.1=\u9047\u5230OB\u81f4\u547d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 5\u5206\u949f,SQLState:{0}
+singletablewritertask.2=\u9047\u5230OB\u53ef\u6062\u590d\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u5206\u949f,SQLState:{0}
+singletablewritertask.3=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u4f11\u7720 1\u79d2,\u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}
+singletablewritertask.4=\u9047\u5230OB\u5f02\u5e38,\u56de\u6eda\u6b64\u6b21\u5199\u5165, \u91c7\u7528\u9010\u6761\u5199\u5165\u63d0\u4ea4,SQLState:{0}multitablewritertask.1=配置的tableList為多表，但未配置分表規則，請檢查您的配置
+multitablewritertask.2=配置的多庫中的表名有重複的，但未配置分庫規則和分表規則，請檢查您的配置
+multitablewritertask.3=配置的所有表名都相同，但未配置分庫規則，請檢查您的配置
+multitablewritertask.4=配置的table和db名稱都相同，此種回流方式不支援
+multitablewritertask.5=列配置資訊有錯誤. 因為您配置的任務中，源頭讀取欄位數:{0}與 目的表要寫入的欄位數:{1}不相等. 請檢查您的配置並作出修改.
+multitablewritertask.6=通過規則計算出來的tableName查找對應的db不存在，tableName={0}, 請檢查您配置的規則.
+multitablewritertask.7=通過規則計算出來的db和table不存在，算出的dbName={0},tableName={1}, 請檢查您配置的規則.
+multitablewritertask.8=通過規則計算出來的db不存在，算出的dbName={0}, 請檢查您配置的規則.
+multitablewritertask.9=通過規則計算出來的dbName[{0}], 存在多張分表，請配置您的分表規則.
+multitablewritertask.10=遇到OB致命異常,回滾此次寫入, 休眠 5分鐘,SQLState:{0}
+multitablewritertask.11=遇到OB可恢復異常,回滾此次寫入, 休眠 1分鐘,SQLState:{0}
+multitablewritertask.12=遇到OB異常,回滾此次寫入, 休眠 1秒,採用逐條寫入提交,SQLState:{0}
+multitablewritertask.13=遇到OB異常,回滾此次寫入, 採用逐條寫入提交,SQLState:{0}
+multitablewritertask.14=寫入表[{0}]失敗,休眠[{1}]毫秒,數據:{2}
+multitablewritertask.15=寫入表[{0}]存在髒數據,record={1}, 寫入異常為:
+
+
+singletablewritertask.1=遇到OB致命異常,回滾此次寫入, 休眠 5分鐘,SQLState:{0}
+singletablewritertask.2=遇到OB可恢復異常,回滾此次寫入, 休眠 1分鐘,SQLState:{0}
+singletablewritertask.3=遇到OB異常,回滾此次寫入, 休眠 1秒,採用逐條寫入提交,SQLState:{0}
+singletablewritertask.4=遇到OB異常,回滾此次寫入, 採用逐條寫入提交,SQLState:{0}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/MultiVersionWriteTask.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/MultiVersionWriteTask.java
new file mode 100644
index 0000000000..4e400b060c
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/MultiVersionWriteTask.java
@@ -0,0 +1,12 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.task;
+
+import com.alibaba.datax.common.util.Configuration;
+
+/**
+ * TODO(yuez)升级hbase api之后再补充暂时用不到
+ */
+public class MultiVersionWriteTask extends ObHBaseWriteTask{
+ public MultiVersionWriteTask(Configuration configuration) throws Exception {
+ super(configuration);
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/NormalWriteTask.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/NormalWriteTask.java
new file mode 100644
index 0000000000..3113c022ce
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/NormalWriteTask.java
@@ -0,0 +1,12 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.task;
+
+import com.alibaba.datax.common.util.Configuration;
+
+/**
+ * TODO(yuez) 升级hbase api之后再补充暂时用不到
+ */
+public class NormalWriteTask extends ObHBaseWriteTask{
+ public NormalWriteTask(Configuration configuration) throws Exception {
+ super(configuration);
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/ObHBaseWriteTask.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/ObHBaseWriteTask.java
new file mode 100644
index 0000000000..d424f6eac2
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/ObHBaseWriteTask.java
@@ -0,0 +1,317 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.task;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.common.util.MessageSource;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Config;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Constant;
+import com.alibaba.datax.plugin.writer.obhbasewriter.NullModeType;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ObHTableInfo;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ext.ServerConnectInfo;
+import com.google.common.collect.Lists;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ObHBaseWriteTask extends CommonRdbmsWriter.Task {
+ private final static MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(ObHBaseWriteTask.class);
+ private final static Logger LOG = LoggerFactory.getLogger(ObHBaseWriteTask.class);
+
+ public NullModeType nullMode = null;
+ private int maxRetryCount;
+
+ public List columns;
+ public List rowkeyColumn;
+ public Configuration versionColumn;
+
+ public String hbaseTableName;
+ public String encoding;
+ public Boolean walFlag;
+
+ String configUrl;
+ String dbName;
+ String ip;
+ String port;
+
+ String fullUserName;
+ boolean usdOdpMode;
+ String sysUsername;
+ String sysPassword;
+ private ObHTableInfo obHTableInfo;
+
+ private ConcurrentTableWriter concurrentWriter;
+ private boolean allTaskInQueue = false;
+ private long startTime = 0;
+ private String threadName = Thread.currentThread().getName();
+
+ private Lock lock = new ReentrantLock();
+ private Condition condition = lock.newCondition();
+
+ public ObHBaseWriteTask(Configuration configuration) {
+ super(DataBaseType.MySql);
+ init(configuration);
+ }
+
+ @Override
+ public void init(com.alibaba.datax.common.util.Configuration configuration) {
+ this.obHTableInfo = new ObHTableInfo(configuration);
+ this.hbaseTableName = configuration.getString(ConfigKey.TABLE);
+ this.columns = configuration.getListConfiguration(ConfigKey.COLUMN);
+ this.rowkeyColumn = configuration.getListConfiguration(ConfigKey.ROWKEY_COLUMN);
+ this.versionColumn = configuration.getConfiguration(ConfigKey.VERSION_COLUMN);
+ this.encoding = configuration.getString(ConfigKey.ENCODING, Constant.DEFAULT_ENCODING);
+ this.nullMode = NullModeType.getByTypeName(configuration.getString(ConfigKey.NULL_MODE, Constant.DEFAULT_NULL_MODE));
+// this.memstoreThreshold = configuration.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD);
+ this.walFlag = configuration.getBool(ConfigKey.WAL_FLAG, true);
+ this.maxRetryCount = configuration.getInt(ConfigKey.MAX_RETRY_COUNT, 3);
+
+ // default 1000 rows are committed together
+ this.batchSize = com.alibaba.datax.plugin.rdbms.writer.Constant.DEFAULT_BATCH_SIZE;
+ this.batchByteSize = com.alibaba.datax.plugin.rdbms.writer.Constant.DEFAULT_BATCH_BYTE_SIZE;
+
+ this.configUrl = configuration.getString(ConfigKey.OBCONFIG_URL);
+ this.jdbcUrl = configuration.getString(ConfigKey.JDBC_URL);
+ this.username = configuration.getString(Key.USERNAME);
+ this.password = configuration.getString(Key.PASSWORD);
+ this.dbName = configuration.getString(Key.DBNAME);
+ this.usdOdpMode = configuration.getBool(ConfigKey.USE_ODP_MODE);
+
+ ServerConnectInfo connectInfo = new ServerConnectInfo(jdbcUrl, username, password);
+ String clusterName = connectInfo.clusterName;
+ this.fullUserName = connectInfo.getFullUserName();
+ final String[] ipPort = connectInfo.ipPort.split(":");
+ if (usdOdpMode) {
+ this.ip = ipPort[0];
+ this.port = ipPort[1];
+ } else {
+ this.sysUsername = configuration.getString(ConfigKey.OB_SYS_USER);
+ this.sysPassword = configuration.getString(ConfigKey.OB_SYS_PASSWORD);
+ connectInfo.setSysUser(sysUsername);
+ connectInfo.setSysPass(sysPassword);
+ if (!configUrl.contains("ObRegion")) {
+ if (configUrl.contains("?")) {
+ configUrl += "&ObRegion=" + clusterName;
+ } else {
+ configUrl += "?ObRegion=" + clusterName;
+ }
+ }
+ if (!configUrl.contains("database")) {
+ configUrl += "&database=" + dbName;
+ }
+ }
+ if (null == concurrentWriter) {
+ concurrentWriter = new ConcurrentTableWriter(configuration, connectInfo);
+ allTaskInQueue = false;
+ }
+ }
+
+ @Override
+ public void prepare(Configuration configuration) {
+ concurrentWriter.start();
+ }
+
+ @Override
+ public void startWrite(RecordReceiver recordReceiver, Configuration configuration, TaskPluginCollector taskPluginCollector) {
+ this.taskPluginCollector = taskPluginCollector;
+ int recordCount = 0;
+ int bufferBytes = 0;
+ List records = new ArrayList<>();
+ try {
+ Record record;
+ while ((record = recordReceiver.getFromReader()) != null) {
+ recordCount++;
+ bufferBytes += record.getMemorySize();
+ records.add(record);
+ // 按照指定的批大小进行批量写入
+ if (records.size() >= batchSize || bufferBytes >= batchByteSize) {
+ concurrentWriter.addBatchRecords(Lists.newArrayList(records));
+ records.clear();
+ bufferBytes = 0;
+ }
+ }
+
+ if (!records.isEmpty()) {
+ concurrentWriter.addBatchRecords(records);
+ }
+ } catch (Throwable e) {
+ LOG.warn("startWrite error unexpected ", e);
+ throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
+ }
+ LOG.info(recordCount + " rows received.");
+ waitTaskFinish();
+ }
+
+ public void waitTaskFinish() {
+ this.allTaskInQueue = true;
+ LOG.info("ConcurrentTableWriter has put all task in queue, queueSize = {}, total = {}, finished = {}",
+ concurrentWriter.getTaskQueueSize(),
+ concurrentWriter.getTotalTaskCount(),
+ concurrentWriter.getFinishTaskCount());
+
+ lock.lock();
+ try {
+ while (!concurrentWriter.checkFinish()) {
+ condition.await(50, TimeUnit.MILLISECONDS);
+ // print statistic
+ LOG.debug("Statistic total task {}, finished {}, queue Size {}",
+ concurrentWriter.getTotalTaskCount(),
+ concurrentWriter.getFinishTaskCount(),
+ concurrentWriter.getTaskQueueSize());
+ concurrentWriter.printStatistics();
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Concurrent table writer wait task finish interrupt");
+ } finally {
+ lock.unlock();
+ }
+ LOG.debug("wait all InsertTask finished ...");
+ }
+
+ public boolean isFinished() {
+ return allTaskInQueue && concurrentWriter.checkFinish();
+ }
+
+ public void singalTaskFinish() {
+ lock.lock();
+ try {
+ condition.signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void collectDirtyRecord(Record record, Throwable throwable) {
+ this.taskPluginCollector.collectDirtyRecord(record, throwable);
+ }
+
+ @Override
+ public void post(Configuration configuration) {
+
+ }
+
+ @Override
+ public void destroy(Configuration configuration) {
+ if (concurrentWriter != null) {
+ concurrentWriter.destory();
+ }
+ super.destroy(configuration);
+ }
+
+ public class ConcurrentTableWriter {
+ private BlockingQueue> queue;
+ private List putTasks;
+ private Configuration config;
+ private AtomicLong totalTaskCount;
+ private AtomicLong finishTaskCount;
+ private ServerConnectInfo connectInfo;
+ private ExecutorService executorService;
+ private final int threadCount;
+
+ public ConcurrentTableWriter(Configuration config, ServerConnectInfo connectInfo) {
+ this.threadCount = config.getInt(Config.WRITER_THREAD_COUNT, Config.DEFAULT_WRITER_THREAD_COUNT);
+ this.queue = new LinkedBlockingQueue>(threadCount << 1);
+ this.putTasks = new ArrayList(threadCount);
+ this.config = config;
+ this.totalTaskCount = new AtomicLong(0);
+ this.finishTaskCount = new AtomicLong(0);
+ this.executorService = Executors.newFixedThreadPool(threadCount);
+ this.connectInfo = connectInfo;
+ }
+
+ public long getTotalTaskCount() {
+ return totalTaskCount.get();
+ }
+
+ public long getFinishTaskCount() {
+ return finishTaskCount.get();
+ }
+
+ public int getTaskQueueSize() {
+ return queue.size();
+ }
+
+ public void increFinishCount() {
+ finishTaskCount.incrementAndGet();
+ }
+
+ // should check after put all the task in the queue
+ public boolean checkFinish() {
+ long finishCount = finishTaskCount.get();
+ long totalCount = totalTaskCount.get();
+ return finishCount == totalCount;
+ }
+
+ public synchronized void start() {
+ for (int i = 0; i < threadCount; ++i) {
+ LOG.info("start {} insert task.", (i + 1));
+ PutTask putTask = new PutTask(threadName, queue, config, connectInfo, obHTableInfo, ObHBaseWriteTask.this);
+ putTask.setWriter(this);
+ putTasks.add(putTask);
+ }
+ for (PutTask task : putTasks) {
+ executorService.execute(task);
+ }
+ }
+
+ public void printStatistics() {
+ long insertTotalCost = 0;
+ long insertTotalCount = 0;
+ for (PutTask task : putTasks) {
+ insertTotalCost += task.getTotalCost();
+ insertTotalCount += task.getPutCount();
+ }
+ long avgCost = 0;
+ if (insertTotalCount != 0) {
+ avgCost = insertTotalCost / insertTotalCount;
+ }
+ ObHBaseWriteTask.LOG.debug("Put {} times, totalCost {} ms, average {} ms",
+ insertTotalCount, insertTotalCost, avgCost);
+ }
+
+ public void addBatchRecords(final List records) throws InterruptedException {
+ boolean isSucc = false;
+ while (!isSucc) {
+ isSucc = queue.offer(records, 5, TimeUnit.MILLISECONDS);
+ }
+ totalTaskCount.incrementAndGet();
+ }
+
+ public synchronized void destory() {
+ if (putTasks != null) {
+ for (PutTask task : putTasks) {
+ task.setStop();
+ task.destroy();
+ }
+ }
+ destroyExecutor();
+ }
+
+ private void destroyExecutor() {
+ if (executorService != null && !executorService.isShutdown()) {
+ executorService.shutdown();
+ try {
+ executorService.awaitTermination(0L, TimeUnit.SECONDS);
+ } catch (InterruptedException var2) {
+ }
+ }
+ }
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/PutTask.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/PutTask.java
new file mode 100644
index 0000000000..768772c0ab
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/task/PutTask.java
@@ -0,0 +1,325 @@
+package com.alibaba.datax.plugin.writer.obhbasewriter.task;
+
+import com.alibaba.datax.common.element.DoubleColumn;
+import com.alibaba.datax.common.element.LongColumn;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.common.util.MessageSource;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ColumnType;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Config;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Hbase094xWriterErrorCode;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ObHTableInfo;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ext.ObHbaseTableHolder;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ext.ServerConnectInfo;
+import com.alipay.oceanbase.hbase.constants.OHConstants;
+import com.alipay.oceanbase.rpc.property.Property;
+
+import com.google.common.base.Stopwatch;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.tuple.Triple;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey.OBHBASE_HTABLE_CLIENT_WRITE_BUFFER;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey.OBHBASE_HTABLE_PUT_WRITE_BUFFER_CHECK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey.TABLE_CLIENT_RPC_EXECUTE_TIMEOUT;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey.WRITE_BUFFER_HIGH_MARK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.ConfigKey.WRITE_BUFFER_LOW_MARK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_HBASE_HTABLE_CLIENT_WRITE_BUFFER;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_NETTY_BUFFER_HIGH_WATERMARK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_NETTY_BUFFER_LOW_WATERMARK;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.Constant.DEFAULT_RPC_EXECUTE_TIMEOUT;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.util.ObHbaseWriterUtils.getColumnByte;
+import static com.alibaba.datax.plugin.writer.obhbasewriter.util.ObHbaseWriterUtils.getRowkey;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_HTABLE_CLIENT_WRITE_BUFFER;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_DATABASE;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_FULL_USER_NAME;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_PARAM_URL;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_PASSWORD;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_SYS_USER_NAME;
+import static com.alipay.oceanbase.hbase.constants.OHConstants.HBASE_OCEANBASE_SYS_PASSWORD;
+
+public class PutTask implements Runnable {
+
+ private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(PutTask.class);
+
+ private static final Logger LOG = LoggerFactory.getLogger(PutTask.class);
+
+ private ObHBaseWriteTask writerTask;
+ private ObHBaseWriteTask.ConcurrentTableWriter writer;
+
+ private long totalCost = 0;
+ private long putCount = 0;
+ private boolean isStop;
+
+ private ObHTableInfo obHTableInfo;
+ private final Configuration versionColumn;
+ // 失败重试次数
+ private final int failTryCount;
+
+ private String parentThreadName;
+ private Queue> queue;
+ private Configuration config;
+ private ServerConnectInfo connInfo;
+
+ private ObHbaseTableHolder tableHolder;
+
+ private final SimpleDateFormat df_second = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private final SimpleDateFormat df_ms = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS");
+
+ public PutTask(String parentThreadName, Queue> recordsQueue, Configuration config, ServerConnectInfo connectInfo, ObHTableInfo obHTableInfo, ObHBaseWriteTask writerTask) {
+ this.parentThreadName = parentThreadName;
+ this.queue = recordsQueue;
+ this.config = config;
+ this.connInfo = connectInfo;
+ this.obHTableInfo = obHTableInfo;
+ this.writerTask = writerTask;
+ this.versionColumn = config.getConfiguration(ConfigKey.VERSION_COLUMN);
+ this.failTryCount = config.getInt(Config.FAIL_TRY_COUNT, Config.DEFAULT_FAIL_TRY_COUNT);
+ this.isStop = false;
+ initTableHolder();
+ }
+
+ private void initTableHolder() {
+ try {
+ org.apache.hadoop.conf.Configuration c = new org.apache.hadoop.conf.Configuration();
+ c.set(HBASE_OCEANBASE_FULL_USER_NAME, writerTask.fullUserName);
+ c.set(HBASE_OCEANBASE_PASSWORD, this.connInfo.password);
+ c.set(HBASE_OCEANBASE_DATABASE, writerTask.dbName);
+ // obkv-table-client is needed the code below
+ if (writerTask.usdOdpMode) {
+ c.setBoolean(OHConstants.HBASE_OCEANBASE_ODP_MODE, true);
+ c.set(OHConstants.HBASE_OCEANBASE_ODP_ADDR, connInfo.host);
+ c.set(OHConstants.HBASE_OCEANBASE_ODP_PORT, connInfo.port);
+ LOG.info("sysUser and sysPassword is empty, build HTABLE in odp mode.");
+ } else {
+ c.set(HBASE_OCEANBASE_PARAM_URL, writerTask.configUrl);
+ c.set(HBASE_OCEANBASE_SYS_USER_NAME, this.connInfo.sysUser);
+ c.set(HBASE_OCEANBASE_SYS_PASSWORD, this.connInfo.sysPass);
+ LOG.info("sysUser and sysPassword is not empty, build HTABLE in sys mode.");
+ }
+ c.set(HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK, config.getString(OBHBASE_HTABLE_PUT_WRITE_BUFFER_CHECK, DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK));
+ c.set(HBASE_HTABLE_CLIENT_WRITE_BUFFER, config.getString(OBHBASE_HTABLE_CLIENT_WRITE_BUFFER, DEFAULT_HBASE_HTABLE_CLIENT_WRITE_BUFFER));
+
+ c.set(Property.RS_LIST_ACQUIRE_CONNECT_TIMEOUT.getKey(), "500");
+ c.set(Property.RS_LIST_ACQUIRE_READ_TIMEOUT.getKey(), "5000");
+ c.set(Property.RPC_EXECUTE_TIMEOUT.getKey(), config.getString(TABLE_CLIENT_RPC_EXECUTE_TIMEOUT, DEFAULT_RPC_EXECUTE_TIMEOUT));
+ c.set(Property.NETTY_BUFFER_LOW_WATERMARK.getKey(), config.getString(WRITE_BUFFER_LOW_MARK, DEFAULT_NETTY_BUFFER_LOW_WATERMARK));
+ c.set(Property.NETTY_BUFFER_HIGH_WATERMARK.getKey(), config.getString(WRITE_BUFFER_HIGH_MARK, DEFAULT_NETTY_BUFFER_HIGH_WATERMARK));
+ this.tableHolder = new ObHbaseTableHolder(c, obHTableInfo.getTableName());
+ } catch (Exception e) {
+ LOG.error("init table holder failed, reason: {}", e.getMessage());
+ throw new IllegalStateException(e);
+ }
+
+ }
+
+ private void batchWrite(final List buffer) {
+ HTableInterface ohTable = null;
+ Stopwatch stopwatch = Stopwatch.createStarted();
+ try {
+ ohTable = this.tableHolder.getOhTable();
+ List puts = buildBatchPutList(buffer);
+ ohTable.put(puts);
+ } catch (Exception e) {
+ if (Objects.isNull(ohTable)) {
+ LOG.error("build obHTable: {} failed. reason: {}", obHTableInfo.getTableName(), e.getMessage());
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.GET_HBASE_TABLE_ERROR, Hbase094xWriterErrorCode.GET_HBASE_TABLE_ERROR.getDescription());
+ }
+ //
+ LOG.error("hbase batch error: " + e);
+ // 出错了之后对该出错的batch逐条重试
+ for (Record record : buffer) {
+ writeOneRecord(ohTable, record);
+ }
+ } finally {
+ this.writer.increFinishCount();
+ putCount++;
+ totalCost += stopwatch.elapsed(TimeUnit.MILLISECONDS);
+ try {
+ if (!Objects.isNull(ohTable)) {
+ ohTable.close();
+ }
+ } catch (Exception e) {
+ LOG.warn("error in closing htable: {}. Reason: {}", obHTableInfo.getFullHbaseTableName(), e.getMessage());
+ }
+ }
+ }
+
+ private void writeOneRecord(HTableInterface ohTable, Record record) {
+ int retryCount = 0;
+ while (retryCount < this.failTryCount) {
+ try {
+ byte[] rowkey = getRowkey(record, obHTableInfo);
+ Put put = new Put(rowkey); // row key
+ boolean hasValidValue = buildPut(put, record);
+
+ if (hasValidValue) {
+ ohTable.put(put);
+ }
+ break;
+ } catch (Exception e) {
+ retryCount++;
+ LOG.error("error in writing: " + e.getMessage() + ", retry count: " + retryCount);
+ if (retryCount == this.failTryCount) {
+ LOG.warn("ERROR : record {}", record);
+ this.writerTask.collectDirtyRecord(record, e);
+ }
+ }
+ }
+ }
+
+ private List buildBatchPutList(List buffer) {
+ List puts = new ArrayList<>();
+ for (Record record : buffer) {
+ byte[] rowkey = getRowkey(record, obHTableInfo);
+ Put put = new org.apache.hadoop.hbase.client.Put(rowkey); // row key
+ boolean hasValidValue = buildPut(put, record);
+ if (hasValidValue) {
+ puts.add(put);
+ }
+ }
+ return puts;
+ }
+
+ private boolean buildPut(Put put, Record record) {
+ boolean hasValidValue = false;
+ long timestamp = buildTimestamp(record);
+ for (Map.Entry> columnInfo : obHTableInfo.getIndexColumnInfoMap().entrySet()) {
+ Integer index = columnInfo.getKey();
+ if (index >= record.getColumnNumber()) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.ILLEGAL_VALUE,
+ MESSAGE_SOURCE.message("normaltask.2", record.getColumnNumber(), index));
+ }
+ ColumnType columnType = columnInfo.getValue().getRight();
+ String familyName = columnInfo.getValue().getLeft();
+ String columnName = columnInfo.getValue().getMiddle();
+
+ byte[] value = getColumnByte(columnType, record.getColumn(index), obHTableInfo);
+ if (value != null) {
+ hasValidValue = true;
+ if (timestamp == -1) {
+ put.add(familyName.getBytes(), // family
+ columnName.getBytes(), // Q
+ value); // V
+ } else {
+ put.add(familyName.getBytes(), // family
+ columnName.getBytes(), // Q
+ timestamp, // timestamp/version
+ value); // V
+ }
+ }
+ }
+
+ return hasValidValue;
+ }
+
+ private long buildTimestamp(Record record) {
+ if (versionColumn == null) {
+ return -1;
+ }
+
+ int index = versionColumn.getInt(ConfigKey.INDEX);
+ long timestamp;
+ if (index == -1) {
+ // user specified the constant as timestamp
+ timestamp = versionColumn.getLong(ConfigKey.VALUE);
+ if (timestamp < 0) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.CONSTRUCT_VERSION_ERROR,
+ MESSAGE_SOURCE.message("normaltask.4"));
+ }
+ } else {
+ // 指定列作为版本,long/doubleColumn直接record.aslong, 其它类型尝试用yyyy-MM-dd HH:mm:ss,
+ // yyyy-MM-dd HH:mm:ss SSS去format
+ if (index >= record.getColumnNumber()) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.CONSTRUCT_VERSION_ERROR,
+ MESSAGE_SOURCE.message("normaltask.5", record.getColumnNumber(), index));
+ }
+
+ if (record.getColumn(index).getRawData() == null) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.CONSTRUCT_VERSION_ERROR,
+ MESSAGE_SOURCE.message("normaltask.6"));
+ }
+
+ if (record.getColumn(index) instanceof LongColumn || record.getColumn(index) instanceof DoubleColumn) {
+ timestamp = record.getColumn(index).asLong();
+ } else {
+ Date date;
+ try {
+ date = df_ms.parse(record.getColumn(index).asString());
+ } catch (ParseException e) {
+ try {
+ date = df_second.parse(record.getColumn(index).asString());
+ } catch (ParseException e1) {
+ LOG.info(MESSAGE_SOURCE.message("normaltask.7", index));
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.CONSTRUCT_VERSION_ERROR, e1);
+ }
+ }
+ timestamp = date.getTime();
+ }
+ }
+
+ return timestamp;
+ }
+
+ public void setStop() {isStop = true;}
+
+ public long getTotalCost() {return totalCost;}
+
+ public long getPutCount() {return putCount;}
+
+ public void destroy() {
+ tableHolder.destroy();
+ }
+
+ void setWriterTask(ObHBaseWriteTask writerTask) {
+ this.writerTask = writerTask;
+ }
+
+ void setWriter(ObHBaseWriteTask.ConcurrentTableWriter writer) {
+ this.writer = writer;
+ }
+
+ @Override
+ public void run() {
+ String currentThreadName = String.format("%s-putTask-%d", parentThreadName, Thread.currentThread().getId());
+ Thread.currentThread().setName(currentThreadName);
+ LOG.debug("Task {} start to execute...", currentThreadName);
+ int sleepTimes = 0;
+ while (!isStop) {
+ try {
+ List records = queue.poll();
+ if (null != records) {
+ batchWrite(records);
+ } else if (writerTask.isFinished()) {
+ writerTask.singalTaskFinish();
+ LOG.debug("not more task, thread exist ...");
+ break;
+ } else {
+ TimeUnit.MILLISECONDS.sleep(5);
+ sleepTimes++;
+ }
+ } catch (InterruptedException e) {
+ LOG.debug("TableWriter is interrupt");
+ } catch (Exception e) {
+ LOG.warn("ERROR UNEXPECTED {}", e);
+ }
+ }
+ LOG.debug("Thread exist...");
+ LOG.debug("sleep {} times, total sleep time: {}", sleepTimes, sleepTimes * 5);
+ }
+}
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_en_US.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_en_US.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_ja_JP.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_ja_JP.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_CN.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_CN.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_HK.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_HK.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_TW.properties b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/LocalStrings_zh_TW.properties
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/ObHbaseWriterUtils.java b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/ObHbaseWriterUtils.java
new file mode 100644
index 0000000000..9ccd3ed84b
--- /dev/null
+++ b/obhbasewriter/src/main/java/com/alibaba/datax/plugin/writer/obhbasewriter/util/ObHbaseWriterUtils.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright (c) 2021 OceanBase ob-loader-dumper is licensed under Mulan PSL v2. You can use this software according to
+ * the terms and conditions of the Mulan PSL v2. You may obtain a copy of Mulan PSL v2 at:
+ *
+ * http://license.coscl.org.cn/MulanPSL2
+ *
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING
+ * BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. See the Mulan PSL v2 for more
+ * details.
+ */
+package com.alibaba.datax.plugin.writer.obhbasewriter.util;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.MessageSource;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ColumnType;
+import com.alibaba.datax.plugin.writer.obhbasewriter.Hbase094xWriterErrorCode;
+import com.alibaba.datax.plugin.writer.obhbasewriter.ObHTableInfo;
+import com.alibaba.datax.plugin.writer.obhbasewriter.task.PutTask;
+import java.nio.charset.Charset;
+import org.apache.commons.lang3.tuple.Triple;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * @author cjyyz
+ * @date 2023/03/23
+ * @since
+ */
+public class ObHbaseWriterUtils {
+
+ private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(PutTask.class);
+
+ public static byte[] getRowkey(Record record, ObHTableInfo obHTableInfo) {
+ byte[] rowkeyBuffer = {};
+ for (Triple rowKeyElement : obHTableInfo.getRowKeyElementList()) {
+ Integer index = rowKeyElement.getLeft();
+ ColumnType columnType = rowKeyElement.getRight();
+ if (index == -1) {
+ String value = rowKeyElement.getMiddle();
+ rowkeyBuffer = Bytes.add(rowkeyBuffer, getValueByte(columnType, value, obHTableInfo.getEncoding()));
+ } else {
+ if (index >= record.getColumnNumber()) {
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.CONSTRUCT_ROWKEY_ERROR, MESSAGE_SOURCE.message("normaltask.3", record.getColumnNumber(), index));
+ }
+ byte[] value = getColumnByte(columnType, record.getColumn(index), obHTableInfo);
+ rowkeyBuffer = Bytes.add(rowkeyBuffer, value);
+ }
+ }
+
+ return rowkeyBuffer;
+ }
+
+ public static byte[] getColumnByte(ColumnType columnType, Column column, ObHTableInfo obHTableInfo) {
+ byte[] bytes;
+ if (column.getRawData() != null && !(columnType == ColumnType.STRING && column.asString().equals("null"))) {
+ switch (columnType) {
+ case INT:
+ bytes = Bytes.toBytes(column.asLong().intValue());
+ break;
+ case LONG:
+ bytes = Bytes.toBytes(column.asLong());
+ break;
+ case DOUBLE:
+ bytes = Bytes.toBytes(column.asDouble());
+ break;
+ case FLOAT:
+ bytes = Bytes.toBytes(column.asDouble().floatValue());
+ break;
+ case SHORT:
+ bytes = Bytes.toBytes(column.asLong().shortValue());
+ break;
+ case BOOLEAN:
+ bytes = Bytes.toBytes(column.asBoolean());
+ break;
+ case STRING:
+ bytes = getValueByte(columnType, column.asString(), obHTableInfo.getEncoding());
+ break;
+ case BINARY:
+ bytes = Bytes.toBytesBinary(column.asString());
+ break;
+ default:
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.ILLEGAL_VALUE, MESSAGE_SOURCE.message("hbaseabstracttask.2", columnType));
+ }
+ } else {
+ switch (obHTableInfo.getNullModeType()) {
+ case Skip:
+ bytes = null;
+ break;
+ case Empty:
+ bytes = HConstants.EMPTY_BYTE_ARRAY;
+ break;
+ default:
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.ILLEGAL_VALUE, MESSAGE_SOURCE.message("hbaseabstracttask.3"));
+ }
+ }
+ return bytes;
+ }
+
+ /**
+ * @param columnType
+ * @param value
+ * @return byte[]
+ */
+ private static byte[] getValueByte(ColumnType columnType, String value, String encoding) {
+ byte[] bytes;
+ if (value != null) {
+ switch (columnType) {
+ case INT:
+ bytes = Bytes.toBytes(Integer.parseInt(value));
+ break;
+ case LONG:
+ bytes = Bytes.toBytes(Long.parseLong(value));
+ break;
+ case DOUBLE:
+ bytes = Bytes.toBytes(Double.parseDouble(value));
+ break;
+ case FLOAT:
+ bytes = Bytes.toBytes(Float.parseFloat(value));
+ break;
+ case SHORT:
+ bytes = Bytes.toBytes(Short.parseShort(value));
+ break;
+ case BOOLEAN:
+ bytes = Bytes.toBytes(Boolean.parseBoolean(value));
+ break;
+ case STRING:
+ bytes = value.getBytes(Charset.forName(encoding));
+ break;
+ default:
+ throw DataXException.asDataXException(Hbase094xWriterErrorCode.ILLEGAL_VALUE, MESSAGE_SOURCE.message("hbaseabstracttask.4", columnType));
+ }
+ } else {
+ bytes = HConstants.EMPTY_BYTE_ARRAY;
+ }
+ return bytes;
+ }
+}
\ No newline at end of file
diff --git a/obhbasewriter/src/main/resources/plugin.json b/obhbasewriter/src/main/resources/plugin.json
new file mode 100644
index 0000000000..6ea96196e4
--- /dev/null
+++ b/obhbasewriter/src/main/resources/plugin.json
@@ -0,0 +1,6 @@
+{
+ "name": "obhbasewriter",
+ "class": "com.alibaba.datax.plugin.writer.obhbasewriter.ObHbaseWriter",
+ "description": "适用于: 生产环境. 原理: TODO",
+ "developer": "alibaba"
+}
diff --git a/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ExecutorTemplate.java b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ExecutorTemplate.java
new file mode 100644
index 0000000000..a027ed7f35
--- /dev/null
+++ b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ExecutorTemplate.java
@@ -0,0 +1,287 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ExecutorTemplate {
+
+ /**
+ * The default thread pool size. Set as the number of available processors by default.
+ */
+ public static int DEFAULT_POOL_SIZE = Runtime.getRuntime().availableProcessors();
+
+ /**
+ * Indicate whether the executor closes automatically.
+ */
+ private final boolean autoClose;
+
+ /**
+ *
+ */
+ private final List> futures;
+
+ /**
+ *
+ */
+ private final ExecutorService internalExecutor;
+
+ private final ExecutorCompletionService completionService;
+
+ /**
+ * Set pool size for ExecutorTemplate.
+ */
+ public static void setPoolSize(int size) {
+ DEFAULT_POOL_SIZE = size;
+ }
+
+ /**
+ * Default: 1024 AutoClose: true
+ *
+ * @param poolName
+ */
+ public ExecutorTemplate(String poolName) {
+ this(defaultExecutor(poolName), true);
+ }
+
+ /**
+ * Default: 1024 AutoClose: true
+ *
+ * @param poolName
+ */
+ public ExecutorTemplate(String poolName, int poolSize) {
+ this(defaultExecutor(poolName, poolSize), true);
+ }
+
+ public ExecutorTemplate(String poolName, int poolSize, boolean autoClose) {
+ this(defaultExecutor(poolName, poolSize), autoClose);
+ }
+
+ /**
+ * Default: 1024
+ *
+ * @param poolName
+ * @param autoClose
+ */
+ public ExecutorTemplate(String poolName, boolean autoClose) {
+ this(defaultExecutor(poolName), autoClose);
+ }
+
+ /**
+ * Default: 1024 AutoClose: true
+ *
+ * @param executor
+ */
+ public ExecutorTemplate(ExecutorService executor) {
+ this(executor, true);
+ }
+
+ /**
+ * @param executor
+ */
+ public ExecutorTemplate(ExecutorService executor, boolean autoClose) {
+ this.autoClose = autoClose;
+ this.internalExecutor = executor;
+ this.completionService = new ExecutorCompletionService<>(executor);
+ this.futures = Collections.synchronizedList(new ArrayList<>());
+ }
+
+ /**
+ * @param poolName
+ * @return ExecutorService
+ */
+ public static ExecutorService defaultExecutor(String poolName) {
+ return defaultExecutor(100000, poolName, DEFAULT_POOL_SIZE);
+ }
+
+ /**
+ * @param poolName
+ * @param poolSize
+ * @return ExecutorService
+ */
+ public static ExecutorService defaultExecutor(String poolName, int poolSize) {
+ return defaultExecutor(100000, poolName, poolSize);
+ }
+
+ /**
+ * @param capacity
+ * @param poolName
+ * @return ExecutorService
+ */
+ public static ExecutorService defaultExecutor(int capacity, String poolName, int poolSize) {
+ return new ThreadPoolExecutor(poolSize, poolSize, 30, TimeUnit.SECONDS, /* */
+ new ArrayBlockingQueue<>(capacity), new NamedThreadFactory(poolName));
+ }
+
+ /**
+ * Submit a callable task
+ *
+ * @param task
+ */
+ public void submit(Callable task) {
+ Future f = this.completionService.submit(task);
+ futures.add(f);
+ check(f);
+ }
+
+ /**
+ * Submit a runnable task
+ *
+ * @param task
+ */
+ public void submit(Runnable task) {
+ Future f = this.completionService.submit(task, null);
+ futures.add(f);
+ check(f);
+ }
+
+ /**
+ * Wait all the task run finished, and get all the results.
+ *
+ * @return List
+ */
+ public List waitForResult() {
+ try {
+ int index = 0;
+ Throwable ex = null;
+ List result = new ArrayList();
+ while (index < futures.size()) {
+ try {
+ Future f = this.completionService.take();
+ result.add(f.get());
+ } catch (Throwable e) {
+ ex = getRootCause(e);
+ break;
+ }
+ index++;
+ }
+ if (ex != null) {
+ cancelAll();
+ throw new RuntimeException(ex);
+ } else {
+ return result;
+ }
+ } finally {
+ clearFutures();
+ if (autoClose) {
+ destroyExecutor();
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public void cancelAll() {
+ for (Future f : futures) {
+ if (!f.isDone() && !f.isCancelled()) {
+ f.cancel(false);
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public void clearFutures() {
+ this.futures.clear();
+ }
+
+ /**
+ *
+ */
+ public void destroyExecutor() {
+ if (internalExecutor != null && !internalExecutor.isShutdown()) {
+ this.internalExecutor.shutdown();
+ try {
+ this.internalExecutor.awaitTermination(0, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+
+ /**
+ * Fast check the future
+ *
+ * @param f
+ */
+ private void check(Future f) {
+ if (f != null && f.isDone()) {
+ try {
+ f.get();
+ } catch (Throwable e) {
+ cancelAll();
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * @param throwable
+ * @return Throwable
+ */
+ private Throwable getRootCause(Throwable throwable) {
+ final Throwable holder = throwable;
+ final List list = new ArrayList<>();
+ while (throwable != null && !list.contains(throwable)) {
+ list.add(throwable);
+ throwable = throwable.getCause();
+ }
+ return list.size() < 2 ? holder : list.get(list.size() - 1);
+ }
+
+ /**
+ * An internal named thread factory
+ */
+ static class NamedThreadFactory implements ThreadFactory {
+
+ /**
+ *
+ */
+ private final boolean daemon;
+
+ /**
+ *
+ */
+ private final String name;
+
+ /**
+ *
+ */
+ private final AtomicInteger seq = new AtomicInteger(0);
+
+ /**
+ * @param name
+ */
+ public NamedThreadFactory(String name) {
+ this(name, false);
+ }
+
+ /**
+ * @param name
+ * @param daemon
+ */
+ public NamedThreadFactory(String name, boolean daemon) {
+ this.name = name;
+ this.daemon = daemon;
+ }
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r);
+ t.setDaemon(daemon);
+ t.setPriority(Thread.NORM_PRIORITY);
+ t.setName((name + seq.incrementAndGet()));
+ return t;
+ }
+ }
+}
\ No newline at end of file
diff --git a/oceanbasev10writer/pom.xml b/oceanbasev10writer/pom.xml
index 11997a1e3f..4f9cbf52ea 100644
--- a/oceanbasev10writer/pom.xml
+++ b/oceanbasev10writer/pom.xml
@@ -28,6 +28,16 @@
com.alibaba.datax
plugin-rdbms-util
${datax-project-version}
+
+
+ guava
+ com.google.guava
+
+
+ com.alibaba
+ druid
+
+
org.slf4j
@@ -43,13 +53,11 @@
4.0.4.RELEASE
test
-
com.alipay.oceanbase
@@ -64,6 +72,19 @@
+
+
+ com.oceanbase
+ oceanbase-client
+ 2.4.11
+
+
+ com.google.guava
+ guava
+
+
+
+
com.oceanbase
shade-ob-partition-calculator
@@ -72,8 +93,13 @@
${pom.basedir}/src/main/libs/shade-ob-partition-calculator-1.0-SNAPSHOT.jar
-
+ com.google.guava
+ guava
+ 27.0-jre
+
+
+
log4j
log4j
1.2.16
@@ -89,6 +115,152 @@
4.11
test
+
+ com.oceanbase
+ obkv-table-client
+ 1.2.6
+
+
+ com.alibaba
+ fastjson
+
+
+
+ org.slf4j
+ slf4j-api
+
+
+
+ com.oceanbase
+ oceanbase-client
+
+
+
+ com.google.guava
+ guava
+
+
+ commons-lang
+ commons-lang
+
+
+ com.alipay.sofa.common
+ sofa-common-tools
+
+
+
+ io.netty
+ netty-codec-dns
+
+
+ io.netty
+ netty-codec-http
+
+
+ io.netty
+ netty-codec-http2
+
+
+ io.netty
+ netty-codec-haproxy
+
+
+ io.netty
+ netty-codec-mqtt
+
+
+ io.netty
+ netty-codec-memcache
+
+
+ io.netty
+ netty-codec-redis
+
+
+ io.netty
+ netty-codec-smtp
+
+
+ io.netty
+ netty-codec-socks
+
+
+ io.netty
+ netty-codec-stomp
+
+
+ io.netty
+ netty-codec-xml
+
+
+
+ io.netty
+ netty-handler-proxy
+
+
+ io.netty
+ netty-handler-ssl-ocsp
+
+
+
+ io.netty
+ netty-resolver-dns
+
+
+ io.netty
+ netty-resolver-dns-classes-macos
+
+
+ io.netty
+ netty-resolver-dns-native-macos
+
+
+
+ io.netty
+ netty-transport-rxtx
+
+
+ io.netty
+ netty-transport-udt
+
+
+ io.netty
+ netty-transport-sctp
+
+
+
+
+ com.alipay.sofa.common
+ sofa-common-tools
+ 1.3.11
+
+
+
+ org.slf4j
+ slf4j-api
+
+
+
+ com.google.guava
+ guava
+
+
+
+
+ com.alibaba
+ fastjson
+ 1.2.83
+
+
+ commons-lang
+ commons-lang
+ 2.6
+
+
+ mysql
+ mysql-connector-java
+ ${mysql.driver.version}
+
diff --git a/package.xml b/package.xml
index e51c11e1d5..624109f799 100644
--- a/package.xml
+++ b/package.xml
@@ -39,6 +39,13 @@
datax
+
+ obhbasereader/target/datax/
+
+ **/*.*
+
+ datax
+
drdsreader/target/datax/
@@ -476,6 +483,13 @@
datax
+
+ obhbasewriter/target/datax/
+
+ **/*.*
+
+ datax
+
gdbwriter/target/datax/
diff --git a/plugin-rdbms-util/pom.xml b/plugin-rdbms-util/pom.xml
index c49f64af16..6dc69e06c6 100755
--- a/plugin-rdbms-util/pom.xml
+++ b/plugin-rdbms-util/pom.xml
@@ -33,6 +33,17 @@
${mysql.driver.version}
test
+
+ com.oceanbase
+ oceanbase-client
+ 2.4.11
+
+
+ com.google.guava
+ guava
+
+
+
org.slf4j
slf4j-api
diff --git a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/util/ObVersion.java b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/util/ObVersion.java
index 0eb34feb0d..da078df924 100644
--- a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/util/ObVersion.java
+++ b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/reader/util/ObVersion.java
@@ -16,6 +16,8 @@ public class ObVersion implements Comparable {
private int patchNumber;
public static final ObVersion V2276 = valueOf("2.2.76");
+ public static final ObVersion V2252 = valueOf("2.2.52");
+ public static final ObVersion V3 = valueOf("3.0.0.0");
public static final ObVersion V4000 = valueOf("4.0.0.0");
private static final ObVersion DEFAULT_VERSION =
diff --git a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/SplitedSlice.java b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/SplitedSlice.java
new file mode 100644
index 0000000000..d8de129490
--- /dev/null
+++ b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/SplitedSlice.java
@@ -0,0 +1,37 @@
+package com.alibaba.datax.plugin.rdbms.util;
+
+public class SplitedSlice {
+ private String begin;
+ private String end;
+ private String range;
+
+ public SplitedSlice(String begin, String end, String range) {
+ this.begin = begin;
+ this.end = end;
+ this.range = range;
+ }
+
+ public String getBegin() {
+ return begin;
+ }
+
+ public void setBegin(String begin) {
+ this.begin = begin;
+ }
+
+ public String getEnd() {
+ return end;
+ }
+
+ public void setEnd(String end) {
+ this.end = end;
+ }
+
+ public String getRange() {
+ return range;
+ }
+
+ public void setRange(String range) {
+ this.range = range;
+ }
+}
diff --git a/pom.xml b/pom.xml
index eeb4bfaffe..c7f43f1725 100644
--- a/pom.xml
+++ b/pom.xml
@@ -55,6 +55,7 @@
oraclereader
cassandrareader
oceanbasev10reader
+ obhbasereader
rdbmsreader
odpsreader
@@ -93,6 +94,7 @@
kingbaseeswriter
adswriter
oceanbasev10writer
+ obhbasewriter
adbpgwriter
hologresjdbcwriter
rdbmswriter
diff --git a/selectdbwriter/doc/stream2selectdb.json b/selectdbwriter/doc/stream2selectdb.json
index d5e14c4884..0c5be6dc0a 100644
--- a/selectdbwriter/doc/stream2selectdb.json
+++ b/selectdbwriter/doc/stream2selectdb.json
@@ -1,92 +1,61 @@
{
- "core":{
- "transport":{
- "channel":{
- "speed":{
- "byte":10485760
+ "core": {
+ "transport": {
+ "channel": {
+ "speed": {
+ "byte": 10485760
}
}
}
},
- "job":{
- "content":[
+ "job": {
+ "content": [
{
- "reader":{
- "name":"streamreader",
- "parameter":{
- "column":[
- {
- "type":"string",
- "value":"DataX"
- },
- {
- "type":"int",
- "value":19890604
- },
- {
- "type":"date",
- "value":"1989-06-04 00:00:00"
- },
- {
- "type":"bool",
- "value":true
- },
- {
- "type":"string",
- "value":"test"
- }
- ],
- "sliceRecordCount":1000000
- }
- },
- "writer":{
- "name":"selectdbwriter",
- "parameter":{
- "loadUrl":[
+ "reader": {},
+ "writer": {
+ "name": "selectdbwriter",
+ "parameter": {
+ "loadUrl": [
"xxx:35871"
],
- "loadProps":{
- "file.type":"json",
- "file.strip_outer_array":"true"
+ "loadProps": {
+ "file.type": "json",
+ "file.strip_outer_array": "true"
},
- "database":"db1",
- "column":[
+ "database": "db1",
+ "column": [
"k1",
"k2",
"k3",
"k4",
"k5"
],
- "username":"admin",
- "password":"SelectDB2022",
- "postSql":[
-
- ],
- "preSql":[
-
- ],
- "connection":[
+ "username": "admin",
+ "password": "SelectDB2022",
+ "postSql": [],
+ "preSql": [],
+ "connection": [
{
- "jdbcUrl":"jdbc:mysql://xxx:32386/cl_test",
- "table":[
+ "jdbcUrl": "jdbc:mysql://xxx:32386/cl_test",
+ "table": [
"test_selectdb"
],
- "selectedDatabase":"cl_test"
+ "selectedDatabase": "cl_test"
}
],
- "maxBatchRows":200000,
- "batchSize":53687091200
+ "maxBatchRows": 200000,
+ "batchSize": 53687091200
}
}
}
],
- "setting":{
- "errorLimit":{
- "percentage":0.02,
- "record":0
+ "setting": {
+ "errorLimit": {
+ "percentage": 0.02,
+ "record": 0
},
- "speed":{
- "byte":10485760
+ "speed": {
+ "byte": 10485760
}
}
}
diff --git a/transformer/doc/transformer.md b/transformer/doc/transformer.md
index 0a00dbaa9c..a9da83a4ed 100644
--- a/transformer/doc/transformer.md
+++ b/transformer/doc/transformer.md
@@ -47,7 +47,7 @@ dx_replace(1,"5","10","****") column 1的value为“dataxTest”=>"datax****"
4. dx_filter （关联filter暂不支持，即多个字段的联合判断，函参太过复杂，用户难以使用。）
* 参数：
* 第一个参数：字段编号，对应record中第几个字段。
- * 第二个参数：运算符，支持一下运算符：like, not like, >, =, <, >=, !=, <=
+ * 第二个参数：运算符，支持以下运算符：like, not like, >, =, <, >=, !=, <=
* 第三个参数：正则表达式（java正则表达式）、值。
* 返回：
* 如果匹配正则表达式，返回Null，表示过滤该行。不匹配表达式时，表示保留该行。（注意是该行）。对于>=<都是对字段直接compare的结果.
@@ -145,11 +145,11 @@ String code3 = "Column column = record.getColumn(1);\n" +
"type": "string"
},
{
- "value": 19890604,
+ "value": 1724154616370,
"type": "long"
},
{
- "value": "1989-06-04 00:00:00",
+ "value": "2024-01-01 00:00:00",
"type": "date"
},
{
@@ -157,11 +157,11 @@ String code3 = "Column column = record.getColumn(1);\n" +
"type": "bool"
},
{
- "value": "test",
+ "value": "TestRawData",
"type": "bytes"
}
],
- "sliceRecordCount": 100000
+ "sliceRecordCount": 100
}
},
"writer": {
@@ -174,38 +174,44 @@ String code3 = "Column column = record.getColumn(1);\n" +
"transformer": [
{
"name": "dx_substr",
- "parameter":
- {
- "columnIndex":5,
- "paras":["1","3"]
- }
+ "parameter": {
+ "columnIndex": 5,
+ "paras": [
+ "1",
+ "3"
+ ]
+ }
},
{
"name": "dx_replace",
- "parameter":
- {
- "columnIndex":4,
- "paras":["3","4","****"]
- }
+ "parameter": {
+ "columnIndex": 4,
+ "paras": [
+ "3",
+ "4",
+ "****"
+ ]
+ }
},
{
"name": "dx_digest",
- "parameter":
- {
- "columnIndex":3,
- "paras":["md5", "toLowerCase"]
- }
+ "parameter": {
+ "columnIndex": 3,
+ "paras": [
+ "md5",
+ "toLowerCase"
+ ]
+ }
},
{
"name": "dx_groovy",
- "parameter":
- {
- "code": "//groovy code//",
- "extraPackage":[
- "import somePackage1;",
- "import somePackage2;"
- ]
- }
+ "parameter": {
+ "code": "//groovy code//",
+ "extraPackage": [
+ "import somePackage1;",
+ "import somePackage2;"
+ ]
+ }
}
]
}