+ * Job类init-->prepare-->split + * + * Task类init-->prepare-->startWrite-->post-->destroy + * Task类init-->prepare-->startWrite-->post-->destroy + * + * Job类post-->destroy + *
+ * Map.put(1, "abc"); + * Map.put(5, "def"); // Error: parameter index is 5, but 2 values exists. + */ + private void checkRange() { + OptionalInt optionalInt = parameters.keySet().stream().mapToInt(e -> e).max(); + int parameterIndex = optionalInt.orElseThrow(() -> new IllegalArgumentException("No parameter index found")); + checkArgument(parameterIndex == parameters.size(), "Parameter index(%s) is unmatched with value list(%s)", parameterIndex, parameters.size()); + } +} \ No newline at end of file diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/directPath/ObTableDirectLoad.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/directPath/ObTableDirectLoad.java new file mode 100644 index 0000000000..859fcedd16 --- /dev/null +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/directPath/ObTableDirectLoad.java @@ -0,0 +1,154 @@ +package com.alibaba.datax.plugin.writer.oceanbasev10writer.directPath; + +import java.sql.SQLException; +import java.util.Objects; + +import com.alipay.oceanbase.rpc.direct_load.ObDirectLoadBucket; +import com.alipay.oceanbase.rpc.direct_load.ObDirectLoadConnection; +import com.alipay.oceanbase.rpc.direct_load.ObDirectLoadStatement; +import com.alipay.oceanbase.rpc.direct_load.ObDirectLoadTraceId; +import com.alipay.oceanbase.rpc.direct_load.exception.ObDirectLoadException; +import com.alipay.oceanbase.rpc.direct_load.protocol.payload.ObTableLoadClientStatus; +import com.alipay.oceanbase.rpc.table.ObTable; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Wrapper of the direct-load API for OceanBase. + */ +public class ObTableDirectLoad implements AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(ObTableDirectLoad.class); + + private final String tableName; + private final String schemaTableName; + private final ObDirectLoadStatement statement; + private final ObDirectLoadConnection connection; + + public ObTableDirectLoad(String schemaName, String tableName, ObDirectLoadStatement statement, ObDirectLoadConnection connection) { + Objects.requireNonNull(schemaName, "schemaName must not be null"); + Objects.requireNonNull(tableName, "tableName must not be null"); + Objects.requireNonNull(statement, "statement must not be null"); + Objects.requireNonNull(connection, "connection must not be null"); + this.tableName = tableName; + this.schemaTableName = String.format("%s.%s", schemaName, tableName); + this.statement = statement; + this.connection = connection; + } + + /** + * Begin the direct load operation. + * + * @throws ObDirectLoadException if an error occurs during the operation. + */ + public void begin() throws ObDirectLoadException { + statement.begin(); + } + + /** + * Write data into the direct load operation. + * + * @param bucket The data bucket to write. + * @throws SQLException if writing fails. + */ + public void write(ObDirectLoadBucket bucket) throws SQLException { + try { + if (bucket == null || bucket.isEmpty()) { + throw new IllegalArgumentException("Bucket must not be null or empty."); + } + LOG.info("Writing {} rows to table: {}", bucket.getRowNum(), schemaTableName); + statement.write(bucket); + LOG.info("Successfully wrote bucket data to table: {}", schemaTableName); + } catch (ObDirectLoadException e) { + LOG.error("Failed to write to table: {}", schemaTableName, e); + throw new SQLException(String.format("Failed to write to table: %s", schemaTableName), e); + } + } + + /** + * Commit the current direct load operation. + * + * @throws SQLException if commit fails. + */ + public void commit() throws SQLException { + try { + LOG.info("Committing direct load for table: {}", schemaTableName); + statement.commit(); + LOG.info("Successfully committed direct load for table: {}", schemaTableName); + } catch (ObDirectLoadException e) { + LOG.error("Failed to commit for table: {}", schemaTableName, e); + throw new SQLException(String.format("Failed to commit for table: %s", schemaTableName), e); + } + } + + /** + * Close the direct load operation. + */ + public void close() { + LOG.info("Closing direct load for table: {}", schemaTableName); + statement.close(); + connection.close(); + LOG.info("Direct load closed for table: {}", schemaTableName); + } + + /** + * Gets the status from the current connection based on the traceId + */ + public ObTableLoadClientStatus getStatus() throws SQLException { + ObDirectLoadTraceId traceId = statement.getTraceId(); + // Check if traceId is null and throw an exception with a clear message + if (traceId == null) { + throw new SQLException("traceId is null."); + } + // Retrieve the status using the traceId + ObTableLoadClientStatus status = statement.getConnection().getProtocol().getHeartBeatRpc(traceId).getStatus(); + if (status == null) { + LOG.info("Direct load connect protocol heartBeatRpc for table is null: {}", schemaTableName); + throw new SQLException("status is null."); + } + // Return status if not null; otherwise, return ERROR + return status; + } + + /** + * Gets the current table + */ + public ObTable getTable() { + try { + return this.statement.getObTablePool().getControlObTable(); + } catch (ObDirectLoadException e) { + throw new RuntimeException(e); + } + } + + public String getTableName() { + if (StringUtils.isBlank(tableName)) { + throw new IllegalArgumentException("tableName is blank."); + } + return tableName; + } + + /** + * Inserts data into the direct load operation. + * + * @param bucket The data bucket containing rows to insert. + * @throws SQLException if an error occurs during the insert operation. + */ + public void insert(ObDirectLoadBucket bucket) throws SQLException { + LOG.info("Inserting {} rows to table: {}", bucket.getRowNum(), schemaTableName); + + if (bucket.isEmpty()) { + LOG.warn("Parameter 'bucket' must not be empty."); + throw new IllegalArgumentException("Parameter 'bucket' must not be empty."); + } + + try { + // Perform the insertion into the load operation + statement.write(bucket); + LOG.info("Successfully inserted data into table: {}", schemaTableName); + } catch (Exception ex) { + LOG.error("Unexpected error during insert operation for table: {}", schemaTableName, ex); + throw new SQLException("Unexpected error during insert operation.", ex); + } + } +} diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/AbstractConnHolder.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/AbstractConnHolder.java index c8630cd0af..6e4d4aab51 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/AbstractConnHolder.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/AbstractConnHolder.java @@ -14,6 +14,17 @@ public abstract class AbstractConnHolder { protected final Configuration config; protected Connection conn; + protected String jdbcUrl; + protected String userName; + protected String password; + + protected AbstractConnHolder(Configuration config, String jdbcUrl, String userName, String password) { + this.config = config; + this.jdbcUrl = jdbcUrl; + this.userName = userName; + this.password = password; + } + public AbstractConnHolder(Configuration config) { this.config = config; } @@ -45,4 +56,6 @@ public Connection reconnect() { public abstract String getUserName(); public abstract void destroy(); + + public abstract void doCommit(); } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathAbstractConnHolder.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathAbstractConnHolder.java new file mode 100644 index 0000000000..c5c6dbe0ca --- /dev/null +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathAbstractConnHolder.java @@ -0,0 +1,61 @@ +package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext; + +import java.sql.Connection; + +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.rdbms.util.DBUtil; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class DirectPathAbstractConnHolder { + private static final Logger LOG = LoggerFactory.getLogger(AbstractConnHolder.class); + protected Configuration config; + protected String jdbcUrl; + protected String userName; + protected String password; + + protected Connection conn; + + protected DirectPathAbstractConnHolder(Configuration config, String jdbcUrl, String userName, String password) { + this.config = config; + this.jdbcUrl = jdbcUrl; + this.userName = userName; + this.password = password; + } + + public Connection reconnect() { + DBUtil.closeDBResources(null, conn); + return initConnection(); + } + + public Connection getConn() { + if (conn == null) { + return initConnection(); + } else { + try { + if (conn.isClosed()) { + return reconnect(); + } + return conn; + } catch (Exception e) { + LOG.debug("can not judge whether the hold connection is closed or not, just reuse the hold connection"); + return conn; + } + } + } + + public String getJdbcUrl() { + return jdbcUrl; + } + + public Configuration getConfig() { + return config; + } + + public void doCommit() {} + + public abstract void destroy(); + + public abstract Connection initConnection(); +} diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathConnHolder.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathConnHolder.java new file mode 100644 index 0000000000..352eda1ca4 --- /dev/null +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DirectPathConnHolder.java @@ -0,0 +1,115 @@ +package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import com.alibaba.datax.common.exception.DataXException; +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.rdbms.util.DBUtil; +import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.common.Table; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.directPath.DirectPathConnection; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObLoadDupActionType; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DirectPathConnHolder extends AbstractConnHolder { + private static final Logger LOG = LoggerFactory.getLogger(DirectPathConnHolder.class); + + /** + * The server side timeout. + */ + private static final long SERVER_TIMEOUT = 24L * 60 * 60 * 1000 * 1000; + + private static final ConcurrentHashMap cache = new ConcurrentHashMap<>(); + + private String tableName; + private String host; + private int rpcPort; + private String tenantName; + private String databaseName; + private int blocks; + private int threads; + private int maxErrors; + private ObLoadDupActionType duplicateKeyAction; + + public DirectPathConnHolder(Configuration config, ServerConnectInfo connectInfo, String tableName, int threadsPerChannel) { + super(config, connectInfo.jdbcUrl, connectInfo.userName, connectInfo.password); + // direct path: + //● publicCloud & odp - single or full + //● publicCloud & observer - not support + //● !publicCloud & odp - full + //● !publicCloud & observer - single + this.userName = connectInfo.getFullUserName(); + this.host = connectInfo.host; + this.rpcPort = connectInfo.rpcPort; + this.tenantName = connectInfo.tenantName; + if (!connectInfo.publicCloud && StringUtils.isEmpty(tenantName)) { + throw new IllegalStateException("tenant name is needed when using direct path load in private cloud."); + } + this.databaseName = connectInfo.databaseName; + this.tableName = tableName; + this.blocks = config.getInt(Config.BLOCKS_COUNT, 1); + this.threads = threadsPerChannel * Math.min(blocks, 32); + this.maxErrors = config.getInt(Config.MAX_ERRORS, 0); + this.duplicateKeyAction = "insert".equalsIgnoreCase(config.getString(Config.OB_WRITE_MODE)) ? ObLoadDupActionType.IGNORE : ObLoadDupActionType.REPLACE; + } + + @Override + public Connection initConnection() { + synchronized (cache) { + conn = cache.computeIfAbsent(new Table(databaseName, tableName), e -> { + try { + return new DirectPathConnection.Builder().host(host) // + .port(rpcPort) // + .tenant(tenantName) // + .user(userName) // + .password(Optional.ofNullable(password).orElse("")) // + .schema(databaseName) // + .table(tableName) // + .blocks(blocks) // + .parallel(threads) // + .maxErrorCount(maxErrors) // + .duplicateKeyAction(duplicateKeyAction) // + .serverTimeout(SERVER_TIMEOUT) // + .configuration(config) + .build(); + } catch (Exception ex) { + throw DataXException.asDataXException(DBUtilErrorCode.CONN_DB_ERROR, ex); + } + }); + } + return conn; + } + + public String getJdbcUrl() { + return ""; + } + + public String getUserName() { + return ""; + } + + @Override + public void destroy() { + if (conn != null && ((DirectPathConnection) conn).isFinished()) { + DBUtil.closeDBResources(null, conn); + } + } + + @Override + public void doCommit() { + try { + if (conn != null) { + conn.commit(); + } + } catch (SQLException e) { + throw new RuntimeException(e); + } + } +} diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java index 262fb1cb7b..02277fbe8c 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java @@ -1,42 +1,54 @@ package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext; import java.sql.Connection; +import java.sql.SQLException; import com.alibaba.datax.common.util.Configuration; /** * wrap oceanbase java client + * * @author oceanbase */ public class OCJConnHolder extends AbstractConnHolder { - private ServerConnectInfo connectInfo; - private String dataSourceKey; - - public OCJConnHolder (Configuration config, ServerConnectInfo connInfo) { - super(config); - this.connectInfo = connInfo; - this.dataSourceKey = OBDataSourceV10.genKey(connectInfo.getFullUserName(), connectInfo.databaseName); - OBDataSourceV10.init(config, connectInfo.getFullUserName(), connectInfo.password, connectInfo.databaseName); - } - - @Override - public Connection initConnection() { - conn = OBDataSourceV10.getConnection(dataSourceKey); - return conn; - } - - @Override - public String getJdbcUrl() { - return connectInfo.jdbcUrl; - } - - @Override - public String getUserName() { - return connectInfo.userName; - } - - public void destroy() { - OBDataSourceV10.destory(this.dataSourceKey); - } + private ServerConnectInfo connectInfo; + private String dataSourceKey; + + public OCJConnHolder(Configuration config, ServerConnectInfo connInfo) { + super(config); + this.connectInfo = connInfo; + this.dataSourceKey = OBDataSourceV10.genKey(connectInfo.getFullUserName(), connectInfo.databaseName); + OBDataSourceV10.init(config, connectInfo.getFullUserName(), connectInfo.password, connectInfo.databaseName); + } + + @Override + public Connection initConnection() { + conn = OBDataSourceV10.getConnection(dataSourceKey); + return conn; + } + + @Override + public String getJdbcUrl() { + return connectInfo.jdbcUrl; + } + + @Override + public String getUserName() { + return connectInfo.userName; + } + + public void destroy() { + OBDataSourceV10.destory(this.dataSourceKey); + } + + public void doCommit() { + try { + if (conn != null) { + conn.commit(); + } + } catch (SQLException e) { + throw new RuntimeException(e); + } + } } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java index ac75d359dc..c0e885c250 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java @@ -1,6 +1,7 @@ package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext; import java.sql.Connection; +import java.sql.SQLException; import java.util.ArrayList; import java.util.List; @@ -14,50 +15,60 @@ * 数据库连接代理对象,负责创建连接，重新连接 * * @author oceanbase - * */ public class ObClientConnHolder extends AbstractConnHolder { - private final String jdbcUrl; - private final String userName; - private final String password; - - public ObClientConnHolder(Configuration config, String jdbcUrl, String userName, String password) { - super(config); - this.jdbcUrl = jdbcUrl; - this.userName = userName; - this.password = password; - } - - // Connect to ob with obclient and obproxy - @Override - public Connection initConnection() { - String BASIC_MESSAGE = String.format("jdbcUrl:[%s]", this.jdbcUrl); - DataBaseType dbType = DataBaseType.OceanBase; - if (ObWriterUtils.isOracleMode()) { - // set up for writing timestamp columns - List sessionConfig = config.getList(Key.SESSION, new ArrayList(), String.class); - sessionConfig.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS'"); - sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF'"); - sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD'"); - config.set(Key.SESSION, sessionConfig); - } - conn = DBUtil.getConnection(dbType, jdbcUrl, userName, password); - DBUtil.dealWithSessionConfig(conn, config, dbType, BASIC_MESSAGE); - return conn; - } - - @Override - public String getJdbcUrl() { - return jdbcUrl; - } - - @Override - public String getUserName() { - return userName; - } - - @Override - public void destroy() { - DBUtil.closeDBResources(null, conn); - } + private final String jdbcUrl; + private final String userName; + private final String password; + + public ObClientConnHolder(Configuration config, String jdbcUrl, String userName, String password) { + super(config); + this.jdbcUrl = jdbcUrl; + this.userName = userName; + this.password = password; + } + + // Connect to ob with obclient and obproxy + @Override + public Connection initConnection() { + String BASIC_MESSAGE = String.format("jdbcUrl:[%s]", this.jdbcUrl); + DataBaseType dbType = DataBaseType.OceanBase; + if (ObWriterUtils.isOracleMode()) { + // set up for writing timestamp columns + List sessionConfig = config.getList(Key.SESSION, new ArrayList(), String.class); + sessionConfig.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS'"); + sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF'"); + sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD'"); + config.set(Key.SESSION, sessionConfig); + } + conn = DBUtil.getConnection(dbType, jdbcUrl, userName, password); + DBUtil.dealWithSessionConfig(conn, config, dbType, BASIC_MESSAGE); + return conn; + } + + @Override + public String getJdbcUrl() { + return jdbcUrl; + } + + @Override + public String getUserName() { + return userName; + } + + @Override + public void destroy() { + DBUtil.closeDBResources(null, conn); + } + + @Override + public void doCommit() { + try { + if (conn != null) { + conn.commit(); + } + } catch (SQLException e) { + throw new RuntimeException(e); + } + } } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java index fe8889e167..714d731296 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java @@ -5,94 +5,112 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; +import com.alibaba.datax.common.util.Configuration; + public class ServerConnectInfo { - - public String clusterName; - public String tenantName; - public String userName; - public String password; - public String databaseName; - public String ipPort; - public String jdbcUrl; - public boolean publicCloud; - /** - * - * @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 = jdbcUrl; - this.password = password; - parseJdbcUrl(jdbcUrl); - parseFullUserName(username); - } + 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 Configuration config; + + public ServerConnectInfo(final String jdbcUrl, final String username, final String password, Configuration config) { + this.jdbcUrl = jdbcUrl; + this.password = password; + this.config = config; + parseJdbcUrl(jdbcUrl); + parseFullUserName(username); + } + + private void parseJdbcUrl(final String jdbcUrl) { + Pattern pattern = Pattern.compile("//([\\w\\.\\-]+:\\d+)/([^\\\\?]*)"); + Matcher matcher = pattern.matcher(jdbcUrl); + if (matcher.find()) { + String ipPort = matcher.group(1); + String dbName = matcher.group(2); + this.ipPort = ipPort; + this.host = ipPort.split(":")[0]; + this.port = ipPort.split(":")[1]; + this.databaseName = dbName; + this.publicCloud = host.endsWith("aliyuncs.com"); + } else { + throw new RuntimeException("Invalid argument:" + jdbcUrl); + } + } - 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; - this.databaseName = dbName; - this.publicCloud = ipPort.split(":")[0].endsWith("aliyuncs.com"); - } else { - throw new RuntimeException("Invalid argument:" + jdbcUrl); - } - } + protected 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); + } + } + } - private void parseFullUserName(final String fullUserName) { - int tenantIndex = fullUserName.indexOf("@"); - int clusterIndex = fullUserName.indexOf("#"); - 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 (!publicCloud || tenantIndex < 0) { - this.userName = tenantIndex < 0 ? fullUserName : fullUserName.substring(0, tenantIndex); - this.clusterName = clusterIndex < 0 ? EMPTY : fullUserName.substring(clusterIndex + 1); - this.tenantName = tenantIndex < 0 ? EMPTY : fullUserName.substring(tenantIndex + 1, clusterIndex); - } else { - // If in public cloud, the username with format user@tenant#cluster should be parsed, otherwise, connection can't be created. - this.userName = fullUserName.substring(0, tenantIndex); - if (clusterIndex > tenantIndex) { - this.tenantName = fullUserName.substring(tenantIndex + 1, clusterIndex); - this.clusterName = fullUserName.substring(clusterIndex + 1); - } else { - this.tenantName = fullUserName.substring(tenantIndex + 1); - this.clusterName = EMPTY; - } - } - } + @Override + public String toString() { + return "ServerConnectInfo{" + + "clusterName='" + clusterName + '\'' + + ", tenantName='" + tenantName + '\'' + + ", userName='" + userName + '\'' + + ", password='" + password + '\'' + + ", databaseName='" + databaseName + '\'' + + ", ipPort='" + ipPort + '\'' + + ", jdbcUrl='" + jdbcUrl + '\'' + + ", host='" + host + '\'' + + ", publicCloud=" + publicCloud + + ", rpcPort=" + rpcPort + + '}'; + } - @Override - public String toString() { - StringBuffer strBuffer = new StringBuffer(); - return strBuffer.append("clusterName:").append(clusterName).append(", tenantName:").append(tenantName) - .append(", userName:").append(userName).append(", databaseName:").append(databaseName) - .append(", ipPort:").append(ipPort).append(", jdbcUrl:").append(jdbcUrl).toString(); - } + public String getFullUserName() { + StringBuilder builder = new StringBuilder(); + builder.append(userName); + if (!EMPTY.equals(tenantName)) { + builder.append("@").append(tenantName); + } - public String getFullUserName() { - StringBuilder builder = new StringBuilder(); - builder.append(userName); - 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(); + } - 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; + } } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/AbstractInsertTask.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/AbstractInsertTask.java new file mode 100644 index 0000000000..d4f215e11d --- /dev/null +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/AbstractInsertTask.java @@ -0,0 +1,127 @@ +package com.alibaba.datax.plugin.writer.oceanbasev10writer.task; + +import java.util.List; +import java.util.Queue; +import java.util.concurrent.TimeUnit; + +import com.alibaba.datax.common.element.Record; +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.AbstractConnHolder; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class AbstractInsertTask implements Runnable { + private static final Logger LOG = LoggerFactory.getLogger(AbstractInsertTask.class); + protected final long taskId; + protected ConcurrentTableWriterTask writerTask; + protected ConcurrentTableWriterTask.ConcurrentTableWriter writer; + protected Queue> queue; + protected boolean isStop; + protected Configuration config; + protected ServerConnectInfo connInfo; + protected AbstractConnHolder connHolder; + protected long totalCost = 0; + protected long insertCount = 0; + private boolean printCost = Config.DEFAULT_PRINT_COST; + private long costBound = Config.DEFAULT_COST_BOUND; + + public AbstractInsertTask(final long taskId, Queue> recordsQueue, Configuration config, ServerConnectInfo connectInfo, ConcurrentTableWriterTask task, ConcurrentTableWriterTask.ConcurrentTableWriter writer) { + this.taskId = taskId; + this.queue = recordsQueue; + this.config = config; + this.connInfo = connectInfo; + this.isStop = false; + this.printCost = config.getBool(Config.PRINT_COST, Config.DEFAULT_PRINT_COST); + this.costBound = config.getLong(Config.COST_BOUND, Config.DEFAULT_COST_BOUND); + this.writer = writer; + this.writerTask = task; + initConnHolder(); + } + + public AbstractInsertTask(final long taskId, Queue> recordsQueue, Configuration config, ServerConnectInfo connectInfo) { + this.taskId = taskId; + this.queue = recordsQueue; + this.config = config; + this.connInfo = connectInfo; + this.isStop = false; + this.printCost = config.getBool(Config.PRINT_COST, Config.DEFAULT_PRINT_COST); + this.costBound = config.getLong(Config.COST_BOUND, Config.DEFAULT_COST_BOUND); + initConnHolder(); + } + + protected abstract void initConnHolder(); + + public void setWriterTask(ConcurrentTableWriterTask writerTask) { + this.writerTask = writerTask; + } + + public void setWriter(ConcurrentTableWriterTask.ConcurrentTableWriter writer) { + this.writer = writer; + } + + private boolean isStop() { + return isStop; + } + + public void setStop() { + isStop = true; + } + + public AbstractConnHolder getConnHolder() { + return connHolder; + } + + public void calStatistic(final long cost) { + writer.increFinishCount(); + insertCount++; + totalCost += cost; + if (this.printCost && cost > this.costBound) { + LOG.info("slow multi insert cost {}ms", cost); + } + } + + @Override + public void run() { + Thread.currentThread().setName(String.format("%d-insertTask-%d", taskId, Thread.currentThread().getId())); + LOG.debug("Task {} start to execute...", taskId); + while (!isStop()) { + try { + List records = queue.poll(); + if (null != records) { + write(records); + } else if (writerTask.isFinished()) { + writerTask.singalTaskFinish(); + LOG.debug("not more task, thread exist ..."); + break; + } else { + TimeUnit.MILLISECONDS.sleep(5); + } + } catch (InterruptedException e) { + LOG.debug("TableWriter is interrupt"); + } catch (Exception e) { + LOG.warn("ERROR UNEXPECTED ", e); + break; + } + } + LOG.debug("Thread exist..."); + } + + protected abstract void write(List records); + + public long getTotalCost() { + return totalCost; + } + + public long getInsertCount() { + return insertCount; + } + + public void destroy() { + if (connHolder != null) { + connHolder.destroy(); + } + } +} diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java index 0ad3a1ed2f..dae31d1ab1 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java @@ -1,5 +1,6 @@ package com.alibaba.datax.plugin.writer.oceanbasev10writer.task; +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.plugin.RecordReceiver; @@ -17,10 +18,13 @@ import com.alibaba.datax.plugin.writer.oceanbasev10writer.part.ObPartitionCalculatorV1; import com.alibaba.datax.plugin.writer.oceanbasev10writer.part.ObPartitionCalculatorV2; import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils; + import com.oceanbase.partition.calculator.enums.ObServerMode; + import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.SQLException; +import java.sql.Types; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -32,8 +36,10 @@ 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; + import static com.alibaba.datax.plugin.writer.oceanbasev10writer.Config.DEFAULT_SLOW_MEMSTORE_THRESHOLD; import static com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils.LoadMode.FAST; import static com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils.LoadMode.PAUSE; @@ -42,15 +48,15 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task { private static final Logger LOG = LoggerFactory.getLogger(ConcurrentTableWriterTask.class); - // memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入 - private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD; - // memstore检查的间隔 - private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND; - // 最后一次检查 - private long lastCheckMemstoreTime; + // memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入 + private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD; + // memstore检查的间隔 + private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND; + // 最后一次检查 + private long lastCheckMemstoreTime; + + private volatile ObWriterUtils.LoadMode loadMode = FAST; - private volatile ObWriterUtils.LoadMode loadMode = FAST; - private static AtomicLong totalTask = new AtomicLong(0); private long taskId = -1; private AtomicBoolean isMemStoreFull = new AtomicBoolean(false); @@ -67,38 +73,41 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task { private String obUpdateColumns = null; private String dbName; private int calPartFailedCount = 0; + private boolean directPath; - public ConcurrentTableWriterTask(DataBaseType dataBaseType) { - super(dataBaseType); - taskId = totalTask.getAndIncrement(); - } - - @Override - public void init(Configuration config) { - super.init(config); - // OceanBase 所有操作都是 insert into on duplicate key update 模式 - // writeMode应该使用enum来定义 - this.writeMode = "update"; + public ConcurrentTableWriterTask(DataBaseType dataBaseType) { + super(dataBaseType); + taskId = totalTask.getAndIncrement(); + } + + @Override + public void init(Configuration config) { + super.init(config); + this.directPath = config.getBool(Config.DIRECT_PATH, false); + // OceanBase 所有操作都是 insert into on duplicate key update 模式 + // writeMode应该使用enum来定义 + this.writeMode = "update"; obWriteMode = config.getString(Config.OB_WRITE_MODE, "update"); - ServerConnectInfo connectInfo = new ServerConnectInfo(jdbcUrl, username, password); - dbName = connectInfo.databaseName; - //init check memstore - this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD); - this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND, - Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND); - - this.connHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl, - connectInfo.getFullUserName(), connectInfo.password); - this.isOracleCompatibleMode = ObWriterUtils.isOracleMode(); - if (isOracleCompatibleMode) { - connectInfo.databaseName = connectInfo.databaseName.toUpperCase(); - //在转义的情况下不翻译 - if (!(table.startsWith("\"") && table.endsWith("\""))) { - table = table.toUpperCase(); - } - - LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s", - connectInfo.databaseName, table)); + ServerConnectInfo connectInfo = new ServerConnectInfo(jdbcUrl, username, password, config); + connectInfo.setRpcPort(config.getInt(Config.RPC_PORT, 0)); + dbName = connectInfo.databaseName; + //init check memstore + this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD); + this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND, + Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND); + + this.connHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl, + connectInfo.getFullUserName(), connectInfo.password); + this.isOracleCompatibleMode = ObWriterUtils.isOracleMode(); + if (isOracleCompatibleMode) { + connectInfo.databaseName = connectInfo.databaseName.toUpperCase(); + //在转义的情况下不翻译 + if (!(table.startsWith("\"") && table.endsWith("\""))) { + table = table.toUpperCase(); + } + + LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s", + connectInfo.databaseName, table)); } if (config.getBool(Config.USE_PART_CALCULATOR, Config.DEFAULT_USE_PART_CALCULATOR)) { @@ -133,37 +142,37 @@ private IObPartCalculator createPartitionCalculator(ServerConnectInfo connectInf return new ObPartitionCalculatorV1(connectInfo, table, columns); } - public boolean isFinished() { - return allTaskInQueue && concurrentWriter.checkFinish(); - } - - public boolean allTaskInQueue() { - return allTaskInQueue; - } - - public void setPutAllTaskInQueue() { - this.allTaskInQueue = true; - LOG.info("ConcurrentTableWriter has put all task in queue, queueSize = {}, total = {}, finished = {}", - concurrentWriter.getTaskQueueSize(), - concurrentWriter.getTotalTaskCount(), - concurrentWriter.getFinishTaskCount()); - } - - private void rewriteSql() { - Connection conn = connHolder.initConnection(); - if (isOracleCompatibleMode && obWriteMode.equalsIgnoreCase("update")) { - // change obWriteMode to insert so the insert statement will be generated. - obWriteMode = "insert"; - } - this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns); - LOG.info("writeRecordSql :{}", this.writeRecordSql); - } + public boolean isFinished() { + return allTaskInQueue && concurrentWriter.checkFinish(); + } + + public boolean allTaskInQueue() { + return allTaskInQueue; + } + + public void setPutAllTaskInQueue() { + this.allTaskInQueue = true; + LOG.info("ConcurrentTableWriter has put all task in queue, queueSize = {}, total = {}, finished = {}", + concurrentWriter.getTaskQueueSize(), + concurrentWriter.getTotalTaskCount(), + concurrentWriter.getFinishTaskCount()); + } + + private void rewriteSql() { + Connection conn = connHolder.initConnection(); + if (isOracleCompatibleMode && obWriteMode.equalsIgnoreCase("update")) { + // change obWriteMode to insert so the insert statement will be generated. + obWriteMode = "insert"; + } + this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns); + LOG.info("writeRecordSql :{}", this.writeRecordSql); + } @Override - public void prepare(Configuration writerSliceConfig) { - super.prepare(writerSliceConfig); - concurrentWriter.start(); - } + public void prepare(Configuration writerSliceConfig) { + super.prepare(writerSliceConfig); + concurrentWriter.start(); + } @Override public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) { @@ -173,25 +182,25 @@ public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCo int retryTimes = 0; boolean needRetry = false; do { - try { - if (retryTimes > 0) { - TimeUnit.SECONDS.sleep((1 << retryTimes)); - DBUtil.closeDBResources(null, connection); - connection = DBUtil.getConnection(dataBaseType, jdbcUrl, username, password); - LOG.warn("getColumnMetaData of table {} failed, retry the {} times ...", this.table, retryTimes); - } - ColumnMetaCache.init(connection, this.table, this.columns); - this.resultSetMetaData = ColumnMetaCache.getColumnMeta(); - needRetry = false; - } catch (SQLException e) { - needRetry = true; - ++retryTimes; - e.printStackTrace(); - LOG.warn("fetch column meta of [{}] failed..., retry {} times", this.table, retryTimes); - } catch (InterruptedException e) { - LOG.warn("startWriteWithConnection interrupt, ignored"); - } finally { - } + try { + if (retryTimes > 0) { + TimeUnit.SECONDS.sleep((1 << retryTimes)); + DBUtil.closeDBResources(null, connection); + connection = DBUtil.getConnection(dataBaseType, jdbcUrl, username, password); + LOG.warn("getColumnMetaData of table {} failed, retry the {} times ...", this.table, retryTimes); + } + ColumnMetaCache.init(connection, this.table, this.columns); + this.resultSetMetaData = ColumnMetaCache.getColumnMeta(); + needRetry = false; + } catch (SQLException e) { + needRetry = true; + ++retryTimes; + e.printStackTrace(); + LOG.warn("fetch column meta of [{}] failed..., retry {} times", this.table, retryTimes); + } catch (InterruptedException e) { + LOG.warn("startWriteWithConnection interrupt, ignored"); + } finally { + } } while (needRetry && retryTimes < 100); try { @@ -200,8 +209,8 @@ public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCo while ((record = recordReceiver.getFromReader()) != null) { if (record.getColumnNumber() != this.columnNumber) { // 源头读取字段列数与目的表字段写入列数不相等，直接报错 - LOG.error("column not equal {} != {}, record = {}", - this.columnNumber, record.getColumnNumber(), record.toString()); + LOG.error("column not equal {} != {}, record = {}", + this.columnNumber, record.getColumnNumber(), record.toString()); throw DataXException .asDataXException( DBUtilErrorCode.CONF_ERROR, @@ -221,231 +230,408 @@ public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCo } } - public PreparedStatement fillStatement(PreparedStatement preparedStatement, Record record) - throws SQLException { - return fillPreparedStatement(preparedStatement, record); - } - - private void addLeftRecords() { - //不需要刷新Cache，已经是最后一批数据了 - for (List groupValues : groupInsertValues.values()) { - if (groupValues.size() > 0 ) { - addRecordsToWriteQueue(groupValues); - } - } - } - - private void addRecordToCache(final Record record) { - Long partId =null; - try { - partId = obPartCalculator == null ? Long.MAX_VALUE : obPartCalculator.calculate(record); - } catch (Exception e1) { - if (calPartFailedCount++ < 10) { - LOG.warn("fail to get partition id: " + e1.getMessage() + ", record: " + record); - } - } + public PreparedStatement fillStatement(PreparedStatement preparedStatement, Record record) + throws SQLException { + return fillPreparedStatement(preparedStatement, record); + } + + private void addLeftRecords() { + //不需要刷新Cache，已经是最后一批数据了 + for (List groupValues : groupInsertValues.values()) { + if (groupValues.size() > 0) { + addRecordsToWriteQueue(groupValues); + } + } + } + + private void addRecordToCache(final Record record) { + Long partId = null; + try { + partId = obPartCalculator == null ? Long.MAX_VALUE : obPartCalculator.calculate(record); + } catch (Exception e1) { + if (calPartFailedCount++ < 10) { + LOG.warn("fail to get partition id: " + e1.getMessage() + ", record: " + record); + } + } if (partId == null) { LOG.debug("fail to calculate parition id, just put into the default buffer."); partId = Long.MAX_VALUE; } - List groupValues = groupInsertValues.computeIfAbsent(partId, k -> new ArrayList(batchSize)); - groupValues.add(record); - if (groupValues.size() >= batchSize) { - groupValues = addRecordsToWriteQueue(groupValues); - groupInsertValues.put(partId, groupValues); - } - } - - /** - * - * @param records - * @return 返回一个新的Cache用于存储接下来的数据 - */ - private List addRecordsToWriteQueue(List records) { - int i = 0; - while (true) { - if (i > 0) { - LOG.info("retry add batch record the {} times", i); - } - try { - concurrentWriter.addBatchRecords(records); - break; - } catch (InterruptedException e) { - i++; - LOG.info("Concurrent table writer is interrupted"); - } - } - return new ArrayList(batchSize); - } - private void checkMemStore() { - Connection checkConn = connHolder.getConn(); - try { - if (checkConn == null || checkConn.isClosed()) { - checkConn = connHolder.reconnect(); - } - }catch (Exception e) { - LOG.warn("Check connection is unusable"); - } - - long now = System.currentTimeMillis(); - if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) { - return; - } - double memUsedRatio = ObWriterUtils.queryMemUsedRatio(checkConn); - if (memUsedRatio >= DEFAULT_SLOW_MEMSTORE_THRESHOLD) { - this.loadMode = memUsedRatio >= memstoreThreshold ? PAUSE : SLOW; - LOG.info("Memstore used ration is {}. Load data {}", memUsedRatio, loadMode.name()); - }else { - this.loadMode = FAST; - } - lastCheckMemstoreTime = now; - } - - public boolean isMemStoreFull() { - return isMemStoreFull.get(); - } - - public boolean isShouldPause() { - return this.loadMode.equals(PAUSE); - } - - public boolean isShouldSlow() { - return this.loadMode.equals(SLOW); - } - - public void print() { - if (LOG.isDebugEnabled()) { - LOG.debug("Statistic total task {}, finished {}, queue Size {}", - concurrentWriter.getTotalTaskCount(), - concurrentWriter.getFinishTaskCount(), - concurrentWriter.getTaskQueueSize()); - concurrentWriter.printStatistics(); - } - } - - public void waitTaskFinish() { - setPutAllTaskInQueue(); - lock.lock(); - try { - while (!concurrentWriter.checkFinish()) { - condition.await(15, TimeUnit.SECONDS); - print(); - checkMemStore(); - } - } catch (InterruptedException e) { - LOG.warn("Concurrent table writer wait task finish interrupt"); - } finally { - lock.unlock(); - } - LOG.debug("wait all InsertTask finished ..."); - } - - public void singalTaskFinish() { - lock.lock(); - condition.signal(); - lock.unlock(); - } - - @Override - public void destroy(Configuration writerSliceConfig) { - if(concurrentWriter!=null) { - concurrentWriter.destory(); - } - // 把本级持有的conn关闭掉 - DBUtil.closeDBResources(null, connHolder.getConn()); - super.destroy(writerSliceConfig); - } - - public class ConcurrentTableWriter { - private BlockingQueue> queue; - private List insertTasks; - private Configuration config; - private ServerConnectInfo connectInfo; - private String rewriteRecordSql; - private AtomicLong totalTaskCount; - private AtomicLong finishTaskCount; - private final int threadCount; - - public ConcurrentTableWriter(Configuration config, ServerConnectInfo connInfo, String rewriteRecordSql) { - threadCount = config.getInt(Config.WRITER_THREAD_COUNT, Config.DEFAULT_WRITER_THREAD_COUNT); - queue = new LinkedBlockingQueue>(threadCount << 1); - insertTasks = new ArrayList(threadCount); - this.config = config; - this.connectInfo = connInfo; - this.rewriteRecordSql = rewriteRecordSql; - this.totalTaskCount = new AtomicLong(0); - this.finishTaskCount = new AtomicLong(0); - } - - 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)); - InsertTask insertTask = new InsertTask(taskId, queue, config, connectInfo, rewriteRecordSql); - insertTask.setWriterTask(ConcurrentTableWriterTask.this); - insertTask.setWriter(this); - insertTasks.add(insertTask); - } - WriterThreadPool.executeBatch(insertTasks); - } - - public void printStatistics() { - long insertTotalCost = 0; - long insertTotalCount = 0; - for (InsertTask task: insertTasks) { - insertTotalCost += task.getTotalCost(); - insertTotalCount += task.getInsertCount(); - } - long avgCost = 0; - if (insertTotalCount != 0) { - avgCost = insertTotalCost / insertTotalCount; - } - ConcurrentTableWriterTask.LOG.debug("Insert {} 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); - checkMemStore(); - } - totalTaskCount.incrementAndGet(); - } - - public synchronized void destory() { - if (insertTasks != null) { - for(InsertTask task : insertTasks) { - task.setStop(); - } - for(InsertTask task: insertTasks) { - task.destroy(); - } - } - } - } + List groupValues = groupInsertValues.computeIfAbsent(partId, k -> new ArrayList(batchSize)); + groupValues.add(record); + if (groupValues.size() >= batchSize) { + groupValues = addRecordsToWriteQueue(groupValues); + groupInsertValues.put(partId, groupValues); + } + } + + /** + * @param records + * @return 返回一个新的Cache用于存储接下来的数据 + */ + private List addRecordsToWriteQueue(List records) { + int i = 0; + while (true) { + if (i > 0) { + LOG.info("retry add batch record the {} times", i); + } + try { + concurrentWriter.addBatchRecords(records); + break; + } catch (InterruptedException e) { + i++; + LOG.info("Concurrent table writer is interrupted"); + } + } + return new ArrayList(batchSize); + } + + private void checkMemStore() { + Connection checkConn = connHolder.getConn(); + try { + if (checkConn == null || checkConn.isClosed()) { + checkConn = connHolder.reconnect(); + } + } catch (Exception e) { + LOG.warn("Check connection is unusable"); + } + + long now = System.currentTimeMillis(); + if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) { + return; + } + double memUsedRatio = ObWriterUtils.queryMemUsedRatio(checkConn); + if (memUsedRatio >= DEFAULT_SLOW_MEMSTORE_THRESHOLD) { + this.loadMode = memUsedRatio >= memstoreThreshold ? PAUSE : SLOW; + LOG.info("Memstore used ration is {}. Load data {}", memUsedRatio, loadMode.name()); + } else { + this.loadMode = FAST; + } + lastCheckMemstoreTime = now; + } + + public boolean isMemStoreFull() { + return isMemStoreFull.get(); + } + + public boolean isShouldPause() { + return this.loadMode.equals(PAUSE); + } + + public boolean isShouldSlow() { + return this.loadMode.equals(SLOW); + } + + public void print() { + if (LOG.isDebugEnabled()) { + LOG.debug("Statistic total task {}, finished {}, queue Size {}", + concurrentWriter.getTotalTaskCount(), + concurrentWriter.getFinishTaskCount(), + concurrentWriter.getTaskQueueSize()); + concurrentWriter.printStatistics(); + } + } + + public void waitTaskFinish() { + setPutAllTaskInQueue(); + lock.lock(); + try { + while (!concurrentWriter.checkFinish()) { + condition.await(15, TimeUnit.SECONDS); + print(); + checkMemStore(); + } + if (directPath){ + concurrentWriter.doCommit(); + } + } catch (InterruptedException e) { + LOG.warn("Concurrent table writer wait task finish interrupt"); + } finally { + lock.unlock(); + } + LOG.debug("wait all InsertTask finished ..."); + } + + public void singalTaskFinish() { + lock.lock(); + condition.signal(); + lock.unlock(); + } + + @Override + public void destroy(Configuration writerSliceConfig) { + if (concurrentWriter != null) { + concurrentWriter.destory(); + } + // 把本级持有的conn关闭掉 + DBUtil.closeDBResources(null, connHolder.getConn()); + super.destroy(writerSliceConfig); + } + + public class ConcurrentTableWriter { + private BlockingQueue> queue; + private List abstractInsertTasks; + private Configuration config; + private ServerConnectInfo connectInfo; + private String rewriteRecordSql; + private AtomicLong totalTaskCount; + private AtomicLong finishTaskCount; + private final int threadCount; + + public ConcurrentTableWriter(Configuration config, ServerConnectInfo connInfo, String rewriteRecordSql) { + threadCount = config.getInt(Config.WRITER_THREAD_COUNT, Config.DEFAULT_WRITER_THREAD_COUNT); + queue = new LinkedBlockingQueue>(threadCount << 1); + abstractInsertTasks = new ArrayList(threadCount); + this.config = config; + this.connectInfo = connInfo; + this.rewriteRecordSql = rewriteRecordSql; + this.totalTaskCount = new AtomicLong(0); + this.finishTaskCount = new AtomicLong(0); + } + + 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)); + AbstractInsertTask insertTask = null; + if (directPath) { + insertTask = new DirectPathInsertTask(taskId, queue, config, connectInfo, ConcurrentTableWriterTask.this, this); + } else { + insertTask = new InsertTask(taskId, queue, config, connectInfo, rewriteRecordSql); + } + insertTask.setWriterTask(ConcurrentTableWriterTask.this); + insertTask.setWriter(this); + abstractInsertTasks.add(insertTask); + } + WriterThreadPool.executeBatch(abstractInsertTasks); + } + + public void doCommit() { + this.abstractInsertTasks.get(0).getConnHolder().doCommit(); + } + + public int getThreadCount() { + return threadCount; + } + + public void printStatistics() { + long insertTotalCost = 0; + long insertTotalCount = 0; + for (AbstractInsertTask task : abstractInsertTasks) { + insertTotalCost += task.getTotalCost(); + insertTotalCount += task.getInsertCount(); + } + long avgCost = 0; + if (insertTotalCount != 0) { + avgCost = insertTotalCost / insertTotalCount; + } + ConcurrentTableWriterTask.LOG.debug("Insert {} 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); + checkMemStore(); + } + totalTaskCount.incrementAndGet(); + } + + public synchronized void destory() { + if (abstractInsertTasks != null) { + for (AbstractInsertTask task : abstractInsertTasks) { + task.setStop(); + } + for (AbstractInsertTask task : abstractInsertTasks) { + task.destroy(); + } + } + } + } + + public String getTable() { + return table; + } + + // 直接使用了两个类变量：columnNumber,resultSetMetaData + protected PreparedStatement fillPreparedStatement(PreparedStatement preparedStatement, Record record) + throws SQLException { + for (int i = 0; i < this.columnNumber; i++) { + int columnSqltype = this.resultSetMetaData.getMiddle().get(i); + String typeName = this.resultSetMetaData.getRight().get(i); + preparedStatement = fillPreparedStatementColumnType(preparedStatement, i, columnSqltype, typeName, record.getColumn(i)); + } + + return preparedStatement; + } + + protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex, + int columnSqltype, String typeName, Column column) throws SQLException { + java.util.Date utilDate; + switch (columnSqltype) { + case Types.CHAR: + case Types.NCHAR: + case Types.CLOB: + case Types.NCLOB: + case Types.VARCHAR: + case Types.LONGVARCHAR: + case Types.NVARCHAR: + case Types.LONGNVARCHAR: + preparedStatement.setString(columnIndex + 1, column + .asString()); + break; + + case Types.SMALLINT: + case Types.INTEGER: + case Types.BIGINT: + case Types.NUMERIC: + case Types.DECIMAL: + case Types.FLOAT: + case Types.REAL: + case Types.DOUBLE: + String strValue = column.asString(); + if (emptyAsNull && "".equals(strValue)) { + preparedStatement.setString(columnIndex + 1, null); + } else { + preparedStatement.setString(columnIndex + 1, strValue); + } + break; + + //tinyint is a little special in some database like mysql {boolean->tinyint(1)} + case Types.TINYINT: + Long longValue = column.asLong(); + if (null == longValue) { + preparedStatement.setString(columnIndex + 1, null); + } else { + preparedStatement.setString(columnIndex + 1, longValue.toString()); + } + break; + + // for mysql bug, see http://bugs.mysql.com/bug.php?id=35115 + case Types.DATE: + if (typeName == null) { + typeName = this.resultSetMetaData.getRight().get(columnIndex); + } + + if (typeName.equalsIgnoreCase("year")) { + if (column.asBigInteger() == null) { + preparedStatement.setString(columnIndex + 1, null); + } else { + preparedStatement.setInt(columnIndex + 1, column.asBigInteger().intValue()); + } + } else { + java.sql.Date sqlDate = null; + try { + utilDate = column.asDate(); + } catch (DataXException e) { + throw new SQLException(String.format( + "Date 类型转换错误：[%s]", column)); + } + + if (null != utilDate) { + sqlDate = new java.sql.Date(utilDate.getTime()); + } + preparedStatement.setDate(columnIndex + 1, sqlDate); + } + break; + + case Types.TIME: + java.sql.Time sqlTime = null; + try { + utilDate = column.asDate(); + } catch (DataXException e) { + throw new SQLException(String.format( + "TIME 类型转换错误：[%s]", column)); + } + + if (null != utilDate) { + sqlTime = new java.sql.Time(utilDate.getTime()); + } + preparedStatement.setTime(columnIndex + 1, sqlTime); + break; + + case Types.TIMESTAMP: + java.sql.Timestamp sqlTimestamp = null; + try { + utilDate = column.asDate(); + } catch (DataXException e) { + throw new SQLException(String.format( + "TIMESTAMP 类型转换错误：[%s]", column)); + } + + if (null != utilDate) { + sqlTimestamp = new java.sql.Timestamp( + utilDate.getTime()); + } + preparedStatement.setTimestamp(columnIndex + 1, sqlTimestamp); + break; + case Types.VARBINARY: + case Types.BLOB: + case Types.LONGVARBINARY: + preparedStatement.setBytes(columnIndex + 1, column + .asBytes()); + break; + case Types.BINARY: + String isArray = column.getRawData().toString(); + if (isArray.startsWith("[") && isArray.endsWith("]")) { + preparedStatement.setString(columnIndex + 1, column + .asString()); + } else { + preparedStatement.setBytes(columnIndex + 1, column + .asBytes()); + } + break; + case Types.BOOLEAN: + preparedStatement.setBoolean(columnIndex + 1, column.asBoolean()); + break; + + // warn: bit(1) -> Types.BIT 可使用setBoolean + // warn: bit(>1) -> Types.VARBINARY 可使用setBytes + case Types.BIT: + if (this.dataBaseType == DataBaseType.MySql) { + preparedStatement.setBoolean(columnIndex + 1, column.asBoolean()); + } else { + preparedStatement.setString(columnIndex + 1, column.asString()); + } + break; + default: + throw DataXException + .asDataXException( + DBUtilErrorCode.UNSUPPORTED_TYPE, + String.format( + "您的配置文件中的列配置信息有误. 因为DataX 不支持数据库写入这种字段类型. 字段名:[%s], 字段类型:[%d], 字段Java类型:[%s]. 请修改表中该字段的类型或者不同步该字段.", + this.resultSetMetaData.getLeft() + .get(columnIndex), + this.resultSetMetaData.getMiddle() + .get(columnIndex), + this.resultSetMetaData.getRight() + .get(columnIndex))); + } + return preparedStatement; + } } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/DirectPathInsertTask.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/DirectPathInsertTask.java new file mode 100644 index 0000000000..185321580e --- /dev/null +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/DirectPathInsertTask.java @@ -0,0 +1,76 @@ +package com.alibaba.datax.plugin.writer.oceanbasev10writer.task; + +import java.text.MessageFormat; +import java.util.Arrays; +import java.util.List; +import java.util.Queue; + +import com.alibaba.datax.common.element.Column; +import com.alibaba.datax.common.element.Column.Type; +import com.alibaba.datax.common.element.Record; +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.common.Table; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.common.TableCache; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.directPath.DirectPathConnection; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.directPath.DirectPathPreparedStatement; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.DirectPathConnHolder; +import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DirectPathInsertTask extends AbstractInsertTask { + private static final Logger LOG = LoggerFactory.getLogger(DirectPathInsertTask.class); + + public DirectPathInsertTask(long taskId, Queue> recordsQueue, Configuration config, ServerConnectInfo connectInfo, ConcurrentTableWriterTask task, ConcurrentTableWriterTask.ConcurrentTableWriter writer) { + super(taskId, recordsQueue, config, connectInfo, task, writer); + } + + @Override + protected void initConnHolder() { + this.connHolder = new DirectPathConnHolder(config, connInfo, writerTask.getTable(), writer.getThreadCount()); + this.connHolder.initConnection(); + } + + @Override + protected void write(List records) { + Table table = TableCache.getInstance().getTable(connInfo.databaseName, writerTask.getTable()); + if (Table.Status.FAILURE.equals(table.getStatus())) { + return; + } + DirectPathConnection conn = (DirectPathConnection) connHolder.getConn(); + if (records != null && !records.isEmpty()) { + long startTime = System.currentTimeMillis(); + try (DirectPathPreparedStatement stmt = conn.createStatement()) { + final int columnNumber = records.get(0).getColumnNumber(); + Object[] values = new Object[columnNumber]; + for (Record record : records) { + for (int i = 0; i < columnNumber; i++) { + Column column = record.getColumn(i); + //处理一下时间类型 + if (column.getType().equals(Type.DATE)) { + values[i] = record.getColumn(i).asString(); + } else { + values[i] = record.getColumn(i).getRawData(); + } + } + stmt.addBatch(values); + } + + int[] result = stmt.executeBatch(); + + if (LOG.isDebugEnabled()) { + LOG.debug("[{}] Insert {} rows success", Thread.currentThread().getName(), Arrays.stream(result).sum()); + } + calStatistic(System.currentTimeMillis() - startTime); + stmt.clearBatch(); + } catch (Throwable ex) { + String msg = MessageFormat.format("Insert data into table \"{0}\" failed. Error: {1}", writerTask.getTable(), ex.getMessage()); + LOG.error(msg, ex); + table.setError(ex); + table.setStatus(Table.Status.FAILURE); + throw new RuntimeException(msg); + } + } + } +} \ No newline at end of file diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java index df80cf7ff2..1524d76f31 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java @@ -20,7 +20,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; -public class InsertTask implements Runnable { +public class InsertTask extends AbstractInsertTask implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(InsertTask.class); @@ -49,6 +49,7 @@ public InsertTask( Configuration config, ServerConnectInfo connectInfo, String writeRecordSql) { + super(taskId, recordsQueue, config, connectInfo); this.taskId = taskId; this.queue = recordsQueue; this.connInfo = connectInfo; @@ -62,11 +63,15 @@ public InsertTask( connHolder.initConnection(); } - void setWriterTask(ConcurrentTableWriterTask writerTask) { + protected void initConnHolder() { + + } + + public void setWriterTask(ConcurrentTableWriterTask writerTask) { this.writerTask = writerTask; } - void setWriter(ConcurrentTableWriter writer) { + public void setWriter(ConcurrentTableWriter writer) { this.writer = writer; } @@ -109,6 +114,10 @@ public void run() { LOG.debug("Thread exist..."); } + protected void write(List records) { + + } + public void destroy() { connHolder.destroy(); } diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java index 8add538259..90657373e9 100644 --- a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java +++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java @@ -4,6 +4,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import org.checkerframework.checker.units.qual.A; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,8 +30,8 @@ public static synchronized void execute(InsertTask task) { executorService.execute(task); } - public static synchronized void executeBatch(List tasks) { - for (InsertTask task : tasks) { + public static synchronized void executeBatch(List tasks) { + for (AbstractInsertTask task : tasks) { executorService.execute(task); } } diff --git a/package.xml b/package.xml index e51c11e1d5..7e122a2705 100644 --- a/package.xml +++ b/package.xml @@ -39,6 +39,13 @@ datax + + obhbasereader/target/datax/ + + **/*.* + + datax + drdsreader/target/datax/ @@ -250,6 +257,13 @@ datax + + gaussdbreader/target/datax/ + + **/*.* + + datax + @@ -476,6 +490,13 @@ datax + + obhbasewriter/target/datax/ + + **/*.* + + datax + gdbwriter/target/datax/ @@ -532,5 +553,19 @@ datax + + gaussdbwriter/target/datax/ + + **/*.* + + datax + + + milvuswriter/target/datax/ + + **/*.* + + 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..1b364a7546 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 @@ -127,6 +129,7 @@ adbmysqlwriter sybasewriter neo4jwriter + milvuswriter plugin-rdbms-util plugin-unstructured-storage-util 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;" + ] + } } ] }