Compare commits

..
Author SHA1 Message Date
dependabot[bot] 5a5731124a Bump fastjson from 1.2.78 to 1.2.83 in /tdenginewriter
Bumps [fastjson](https://github.com/alibaba/fastjson) from 1.2.78 to 1.2.83.
- [Release notes](https://github.com/alibaba/fastjson/releases)
- [Commits](https://github.com/alibaba/fastjson/compare/1.2.78...1.2.83)

---
updated-dependencies:
- dependency-name: com.alibaba:fastjson
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
2022-06-17 03:38:37 +00:00
Trafalgar 894b57bd3d Merge pull request #1318 from johnrobbet/obreader_read_by_partition
Obreader read by partition
2022-06-16 21:26:42 +08:00
Trafalgar 8c92ce2caa Update README.md
钉钉群容量受到管控,暂时先隐藏过期分享二维码
2022-06-13 20:55:41 +08:00
Trafalgar ab6cf73527 Update README.md
更新二进制包datax.tar.gz下载地址,为最新版本构建包地址
2022-06-13 16:45:14 +08:00
Trafalgar 47a9ac5642 Update README.md
更新README文档,介绍月度迭代机制和datax_v202205迭代内容
2022-06-11 12:37:45 +08:00
johnrobbet e740300481 do not change columnName to upper case when searching for save point columns 2022-03-11 16:01:02 +08:00
johnrobbet 04c3c2e810 do not readBySplit when it's not in table mode 2022-03-11 16:01:02 +08:00
johnrobbet da9134bda4 readByPartition is lower priority than splitPk 2022-03-11 16:01:02 +08:00
johnrobbet 8d9e936047 obreader: support table name with schema name 2022-03-11 16:01:02 +08:00
johnrobbet f6f94821bc fix number type splitPk in ob oracle mode 2022-03-11 16:01:02 +08:00
johnrobbet d1a17c5c91 fix where missing in single table split 2022-03-11 16:01:02 +08:00
johnrobbet 1a5f1f0288 escape the keywords for splitPk 2022-03-11 16:01:02 +08:00
江煦 2d37cf10b9 obreader: readByPartition 2022-03-11 16:01:01 +08:00
Jing Xu cb518114dc Merge branch 'alibaba:master' into obreader_read_by_partition 2022-03-11 15:55:06 +08:00
江煦 a02a69f2bd dbname with minus('-') can not be parsed in ob writer 2021-09-18 19:54:49 +08:00
13 changed files with 482 additions and 59 deletions
+11 -16
View File
@@ -25,7 +25,8 @@ DataX本身作为数据同步框架,将不同数据源的同步抽象为从源
# Quick Start
##### Download [DataX下载地址](http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz)
##### Download [DataX下载地址](https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/20220530/datax.tar.gz)
##### 请点击:[Quick Start](https://github.com/alibaba/DataX/blob/master/userGuid.md)
@@ -90,6 +91,13 @@ DataX目前已经有了比较全面的插件体系,主流的RDBMS数据库、N
请点击:[DataX插件开发宝典](https://github.com/alibaba/DataX/blob/master/dataxPluginDev.md)
# 重要版本更新说明
DataX 后续计划月度迭代更新,也欢迎感兴趣的同学提交 Pull requests,月度更新内容会介绍介绍如下。
- [datax_v202205]https://github.com/alibaba/DataX/releases/tag/datax_v202205)
- 涉及通道能力更新(MaxCompute、Hologres、OSS、Tdengine等)、安全漏洞更新、通用打包更新等
# 项目成员
@@ -137,23 +145,10 @@ This software is free to use under the Apache License [Apache license](https://g
8. 对高并发、高稳定可用性、高性能、大数据处理有过实际项目及产品经验者优先考虑;
9. 有大数据产品、云产品、中间件技术解决方案者优先考虑。
````
钉钉用户群:
- DataX开源用户交流群
- <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A4.jpg" width="20%" height="20%">
用户咨询支持:
- DataX开源用户交流群2
- <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A42.jpg" width="20%" height="20%">
钉钉群目前暂时受到了一些管控策略影响,建议大家有问题优先在这里提交问题 Issue,DataX研发和社区会定期回答Issue中的问题,知识库丰富后也能帮助到后来的使用者。
- DataX开源用户交流群3
- <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A43.jpg" width="20%" height="20%">
- DataX开源用户交流群4
- <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A44.jpg" width="20%" height="20%">
- DataX开源用户交流群5
- <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A45.jpg" width="20%" height="20%">
- DataX开源用户交流群6
- <img src="https://user-images.githubusercontent.com/1905000/124073771-139cbd00-da75-11eb-9a3f-598cba145a76.png" width="20%" height="20%">
@@ -3,6 +3,7 @@ package com.alibaba.datax.plugin.reader.oceanbasev10reader;
import java.sql.Connection;
import java.util.List;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ObReaderKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +53,21 @@ public class OceanBaseReader extends Reader {
@Override
public List<Configuration> split(int adviceNumber) {
String splitPk = originalConfig.getString(Key.SPLIT_PK);
List<String> quotedColumns = originalConfig.getList(Key.COLUMN_LIST, String.class);
if (splitPk != null && splitPk.length() > 0 && quotedColumns != null) {
String escapeChar = ObReaderUtils.isOracleMode(originalConfig.getString(ObReaderKey.OB_COMPATIBILITY_MODE))
? "\"" : "`";
if (!splitPk.startsWith(escapeChar) && !splitPk.endsWith(escapeChar)) {
splitPk = escapeChar + splitPk + escapeChar;
}
for (String column : quotedColumns) {
if (column.equals(splitPk)) {
LOG.info("splitPk is an ob reserved keyword, set to {}", splitPk);
originalConfig.set(Key.SPLIT_PK, splitPk);
}
}
}
return this.readerJob.split(this.originalConfig, adviceNumber);
}
@@ -86,6 +102,7 @@ public class OceanBaseReader extends Reader {
String obJdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:");
Connection conn = DBUtil.getConnection(DataBaseType.OceanBase, obJdbcUrl, username, password);
String compatibleMode = ObReaderUtils.getCompatibleMode(conn);
config.set(ObReaderKey.OB_COMPATIBILITY_MODE, compatibleMode);
if (ObReaderUtils.isOracleMode(compatibleMode)) {
ObReaderUtils.compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
}
@@ -0,0 +1,11 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
/**
* @author johnrobbet
*/
public class Constant {
public static String WEAK_READ_QUERY_SQL_TEMPLATE_WITHOUT_WHERE = "select /*+read_consistency(weak)*/ %s from %s ";
public static String WEAK_READ_QUERY_SQL_TEMPLATE = "select /*+read_consistency(weak)*/ %s from %s where (%s)";
}
@@ -0,0 +1,16 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
/**
* @author johnrobbet
*/
public class ObReaderKey {
public final static String READ_BY_PARTITION = "readByPartition";
public final static String PARTITION_NAME = "partitionName";
public final static String PARTITION_TYPE = "partitionType";
public final static String OB_COMPATIBILITY_MODE = "obCompatibilityMode";
}
@@ -1,15 +1,16 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
import java.util.Arrays;
import java.util.List;
import com.alibaba.datax.common.constant.CommonConstant;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
import com.alibaba.datax.plugin.rdbms.reader.Key;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.rdbms.writer.Constant;
import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.PartitionSplitUtil;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,37 +30,62 @@ public class ReaderJob extends CommonRdbmsReader.Job {
ObReaderUtils.escapeDatabaseKeywords(columns);
originalConfig.set(Key.COLUMN, columns);
List<JSONObject> conns = originalConfig.getList(com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, JSONObject.class);
List<JSONObject> conns = originalConfig.getList(Constant.CONN_MARK, JSONObject.class);
for (int i = 0; i < conns.size(); i++) {
JSONObject conn = conns.get(i);
Configuration connConfig = Configuration.from(conn.toString());
List<String> tables = connConfig.getList(Key.TABLE, String.class);
ObReaderUtils.escapeDatabaseKeywords(tables);
originalConfig.set(String.format("%s[%d].%s", com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, i, Key.TABLE), tables);
// tables will be null when querySql is configured
if (tables != null) {
ObReaderUtils.escapeDatabaseKeywords(tables);
originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE),
tables);
}
}
super.init(originalConfig);
}
@Override
public List<Configuration> split(Configuration originalConfig, int adviceNumber) {
List<Configuration> list = super.split(originalConfig, adviceNumber);
List<Configuration> list;
// readByPartition is lower priority than splitPk.
// and readByPartition only works in table mode.
if (!isSplitPkValid(originalConfig) &&
originalConfig.getBool(Constant.IS_TABLE_MODE) &&
originalConfig.getBool(ObReaderKey.READ_BY_PARTITION, false)) {
LOG.info("try to split reader job by partition.");
list = PartitionSplitUtil.splitByPartition(originalConfig);
} else {
LOG.info("try to split reader job by splitPk.");
list = super.split(originalConfig, adviceNumber);
}
for (Configuration config : list) {
String jdbcUrl = config.getString(Key.JDBC_URL);
String obRegionName = getObRegionName(jdbcUrl);
config.set(CommonConstant.LOAD_BALANCE_RESOURCE_MARK, obRegionName);
}
return list;
}
private boolean isSplitPkValid(Configuration originalConfig) {
String splitPk = originalConfig.getString(Key.SPLIT_PK);
return splitPk != null && splitPk.trim().length() > 0;
}
private String getObRegionName(String jdbcUrl) {
if (jdbcUrl.startsWith(Constant.OB10_SPLIT_STRING)) {
String[] ss = jdbcUrl.split(Constant.OB10_SPLIT_STRING_PATTERN);
final String obJdbcDelimiter = com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING;
if (jdbcUrl.startsWith(obJdbcDelimiter)) {
String[] ss = jdbcUrl.split(obJdbcDelimiter);
if (ss.length >= 2) {
String tenant = ss[1].trim();
String[] sss = tenant.split(":");
return sss[0];
}
}
return null;
}
}
File diff suppressed because one or more lines are too long
@@ -0,0 +1,35 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
import java.util.ArrayList;
import java.util.List;
/**
* @author johnrobbet
*/
public class PartInfo {
private PartType partType;
List<String> partList;
public PartInfo(PartType partType) {
this.partType = partType;
this.partList = new ArrayList();
}
public String getPartType () {
return partType.getTypeString();
}
public void addPart(List partList) {
this.partList.addAll(partList);
}
public List<String> getPartList() {
return partList;
}
public boolean isPartitionTable() {
return partType != PartType.NONPARTITION && partList.size() > 0;
}
}
@@ -0,0 +1,23 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
/**
* @author johnrobbet
*/
public enum PartType {
NONPARTITION("NONPARTITION"),
PARTITION("PARTITION"),
SUBPARTITION("SUBPARTITION");
private String typeString;
PartType (String typeString) {
this.typeString = typeString;
}
public String getTypeString() {
return typeString;
}
}
@@ -0,0 +1,165 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.rdbms.reader.Key;
import com.alibaba.datax.plugin.rdbms.reader.util.HintUtil;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ObReaderKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
/**
* @author johnrobbet
*/
public class PartitionSplitUtil {
private static final Logger LOG = LoggerFactory.getLogger(PartitionSplitUtil.class);
public static List<Configuration> splitByPartition (Configuration configuration) {
List<Configuration> allSlices = new ArrayList<>();
List<Object> conns = configuration.getList(Constant.CONN_MARK, Object.class);
for (int i = 0, len = conns.size(); i < len; i++) {
Configuration sliceConfig = configuration.clone();
Configuration connConf = Configuration.from(conns.get(i).toString());
String jdbcUrl = connConf.getString(Key.JDBC_URL);
sliceConfig.set(Key.JDBC_URL, jdbcUrl);
sliceConfig.remove(Constant.CONN_MARK);
List<String> tables = connConf.getList(Key.TABLE, String.class);
for (String table : tables) {
Configuration tempSlice = sliceConfig.clone();
tempSlice.set(Key.TABLE, table);
allSlices.addAll(splitSinglePartitionTable(tempSlice));
}
}
return allSlices;
}
private static List<Configuration> splitSinglePartitionTable(Configuration configuration) {
String table = configuration.getString(Key.TABLE);
String where = configuration.getString(Key.WHERE, null);
String column = configuration.getString(Key.COLUMN);
final boolean weakRead = configuration.getBool(Key.WEAK_READ, true);
List<Configuration> slices = new ArrayList();
PartInfo partInfo = getObPartInfoBySQL(configuration, table);
if (partInfo != null && partInfo.isPartitionTable()) {
String partitionType = partInfo.getPartType();
for (String partitionName : partInfo.getPartList()) {
LOG.info(String.format("add %s %s for table %s", partitionType, partitionName, table));
Configuration slice = configuration.clone();
slice.set(ObReaderKey.PARTITION_NAME, partitionName);
slice.set(ObReaderKey.PARTITION_TYPE, partitionType);
slice.set(Key.QUERY_SQL,
ObReaderUtils.buildQuerySql(weakRead, column,
String.format("%s partition(%s)", table, partitionName), where));
slices.add(slice);
}
} else {
LOG.info("fail to get table part info or table is not partitioned, proceed as non-partitioned table.");
Configuration slice = configuration.clone();
slice.set(Key.QUERY_SQL, ObReaderUtils.buildQuerySql(weakRead, column, table, where));
slices.add(slice);
}
return slices;
}
private static PartInfo getObPartInfoBySQL(Configuration config, String table) {
PartInfo partInfo = new PartInfo(PartType.NONPARTITION);
List<String> partList;
Connection conn = null;
try {
String jdbcUrl = config.getString(Key.JDBC_URL);
String username = config.getString(Key.USERNAME);
String password = config.getString(Key.PASSWORD);
String dbname = ObReaderUtils.getDbNameFromJdbcUrl(jdbcUrl);
String allTable = "__all_table";
conn = DBUtil.getConnection(DataBaseType.OceanBase, jdbcUrl, username, password);
String obVersion = getResultsFromSql(conn, "select version()").get(0);
LOG.info("obVersion: " + obVersion);
if (ObReaderUtils.compareObVersion("2.2.76", obVersion) < 0) {
allTable = "__all_table_v2";
}
String queryPart = String.format(
"select p.part_name " +
"from oceanbase.__all_part p, oceanbase.%s t, oceanbase.__all_database d " +
"where p.table_id = t.table_id " +
"and d.database_id = t.database_id " +
"and d.database_name = '%s' " +
"and t.table_name = '%s'", allTable, dbname, table);
String querySubPart = String.format(
"select p.sub_part_name " +
"from oceanbase.__all_sub_part p, oceanbase.%s t, oceanbase.__all_database d " +
"where p.table_id = t.table_id " +
"and d.database_id = t.database_id " +
"and d.database_name = '%s' " +
"and t.table_name = '%s'", allTable, dbname, table);
if (config.getString(ObReaderKey.OB_COMPATIBILITY_MODE).equals("ORACLE")) {
queryPart = String.format(
"select partition_name from all_tab_partitions where TABLE_OWNER = '%s' and table_name = '%s'",
dbname.toUpperCase(), table.toUpperCase());
querySubPart = String.format(
"select subpartition_name from all_tab_subpartitions where TABLE_OWNER = '%s' and table_name = '%s'",
dbname.toUpperCase(), table.toUpperCase());
}
PartType partType = PartType.SUBPARTITION;
// try subpartition first
partList = getResultsFromSql(conn, querySubPart);
// if table is not sub-partitioned, the try partition
if (partList.isEmpty()) {
partList = getResultsFromSql(conn, queryPart);
partType = PartType.PARTITION;
}
if (!partList.isEmpty()) {
partInfo = new PartInfo(partType);
partInfo.addPart(partList);
}
} catch (Exception ex) {
LOG.error("error when get partition list: " + ex.getMessage());
} finally {
DBUtil.closeDBResources(null, conn);
}
return partInfo;
}
private static List<String> getResultsFromSql(Connection conn, String sql) {
List<String> list = new ArrayList();
Statement stmt = null;
ResultSet rs = null;
LOG.info("executing sql: " + sql);
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
while (rs.next()) {
list.add(rs.getString(1));
}
} catch (Exception e) {
LOG.error("error when executing sql: " + e.getMessage());
} finally {
DBUtil.closeDBResources(rs, stmt, null);
}
return list;
}
}
@@ -19,15 +19,6 @@ public class TaskContext {
private boolean weakRead = true;
private String userSavePoint;
private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
public String getPartitionName() {
return partitionName;
}
public void setPartitionName(String partitionName) {
this.partitionName = partitionName;
}
private String partitionName;
// 断点续读的保存点
@@ -174,4 +165,12 @@ public class TaskContext {
public void setCompatibleMode(String compatibleMode) {
this.compatibleMode = compatibleMode;
}
public String getPartitionName() {
return partitionName;
}
public void setPartitionName(String partitionName) {
this.partitionName = partitionName;
}
}
@@ -0,0 +1,22 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
import org.junit.Test;
public class ObReaderUtilsTest {
@Test
public void getDbTest() {
assert ObReaderUtils.getDbNameFromJdbcUrl("jdbc:mysql://127.0.0.1:3306/testdb").equalsIgnoreCase("testdb");
assert ObReaderUtils.getDbNameFromJdbcUrl("jdbc:oceanbase://127.0.0.1:2883/testdb").equalsIgnoreCase("testdb");
assert ObReaderUtils.getDbNameFromJdbcUrl("||_dsc_ob10_dsc_||obcluster:mysql||_dsc_ob10_dsc_||jdbc:mysql://127.0.0.1:3306/testdb").equalsIgnoreCase("testdb");
assert ObReaderUtils.getDbNameFromJdbcUrl("||_dsc_ob10_dsc_||obcluster:oracle||_dsc_ob10_dsc_||jdbc:oceanbase://127.0.0.1:3306/testdb").equalsIgnoreCase("testdb");
}
@Test
public void compareObVersionTest() {
assert ObReaderUtils.compareObVersion("2.2.70", "3.2.2") == -1;
assert ObReaderUtils.compareObVersion("2.2.70", "2.2.50") == 1;
assert ObReaderUtils.compareObVersion("2.2.70", "3.1.2") == -1;
assert ObReaderUtils.compareObVersion("3.1.2", "3.1.2") == 0;
}
}
@@ -93,6 +93,7 @@ public class SingleTableSplitUtil {
allQuerySql.add(tempQuerySql);
tempConfig.set(Key.QUERY_SQL, tempQuerySql);
tempConfig.set(Key.WHERE, (hasWhere ? ("(" + where + ") and") : "") + range);
pluginParams.add(tempConfig);
}
} else {
@@ -103,6 +104,7 @@ public class SingleTableSplitUtil {
+ String.format(" %s IS NOT NULL", splitPkName);
allQuerySql.add(tempQuerySql);
tempConfig.set(Key.QUERY_SQL, tempQuerySql);
tempConfig.set(Key.WHERE, (hasWhere ? "(" + where + ") and" : "") + String.format(" %s IS NOT NULL", splitPkName));
pluginParams.add(tempConfig);
}
@@ -118,6 +120,7 @@ public class SingleTableSplitUtil {
StringUtils.join(allQuerySql, "\n"));
tempConfig.set(Key.QUERY_SQL, tempQuerySql);
tempConfig.set(Key.WHERE, (hasWhere ? "(" + where + ") and" : "") + String.format(" %s IS NULL", splitPkName));
pluginParams.add(tempConfig);
return pluginParams;
@@ -254,6 +257,7 @@ public class SingleTableSplitUtil {
switch (SingleTableSplitUtil.DATABASE_TYPE) {
case Oracle:
case OceanBase:
isValidLongType |= type == Types.NUMERIC;
break;
default:
+1 -1
View File
@@ -23,7 +23,7 @@
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.78</version>
<version>1.2.83</version>
</dependency>
<dependency>