Skip to content

D
DataX
Commit a57425d5 authored by chase.wc's avatar chase.wc
Browse files
Options

Write AnalyticDB by adbclient

parent d4d1ea6a
Hide whitespace changes
Inline Side-by-side
Showing with 339 additions and 59 deletions
adswriter/pom.xml
View file @ a57425d5
...@@ -39,6 +39,22 @@ ...@@ -39,6 +39,22 @@
<groupId>com.alibaba.datax</groupId> <groupId>com.alibaba.datax</groupId>
<artifactId>plugin-rdbms-util</artifactId> <artifactId>plugin-rdbms-util</artifactId>
<version>${datax-project-version}</version> <version>${datax-project-version}</version>
<exclusions>
<exclusion>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.alibaba.cloud.analyticdb</groupId>
<artifactId>adbclient</artifactId>
<version>1.0.2</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.12</version>
</dependency> </dependency>
<dependency> <dependency>
......
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/AdsWriter.java
View file @ a57425d5
...@@ -10,8 +10,10 @@ import com.alibaba.datax.plugin.rdbms.util.DataBaseType; ...@@ -10,8 +10,10 @@ import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil; import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
import com.alibaba.datax.plugin.writer.adswriter.ads.ColumnInfo; import com.alibaba.datax.plugin.writer.adswriter.ads.ColumnInfo;
import com.alibaba.datax.plugin.writer.adswriter.ads.TableInfo; import com.alibaba.datax.plugin.writer.adswriter.ads.TableInfo;
import com.alibaba.datax.plugin.writer.adswriter.insert.AdsClientProxy;
import com.alibaba.datax.plugin.writer.adswriter.insert.AdsInsertProxy; import com.alibaba.datax.plugin.writer.adswriter.insert.AdsInsertProxy;
import com.alibaba.datax.plugin.writer.adswriter.insert.AdsInsertUtil; import com.alibaba.datax.plugin.writer.adswriter.insert.AdsInsertUtil;
import com.alibaba.datax.plugin.writer.adswriter.insert.AdsProxy;
import com.alibaba.datax.plugin.writer.adswriter.load.AdsHelper; import com.alibaba.datax.plugin.writer.adswriter.load.AdsHelper;
import com.alibaba.datax.plugin.writer.adswriter.load.TableMetaHelper; import com.alibaba.datax.plugin.writer.adswriter.load.TableMetaHelper;
import com.alibaba.datax.plugin.writer.adswriter.load.TransferProjectConf; import com.alibaba.datax.plugin.writer.adswriter.load.TransferProjectConf;
...@@ -168,8 +170,9 @@ public class AdsWriter extends Writer { ...@@ -168,8 +170,9 @@ public class AdsWriter extends Writer {
userConfiguredPartitions = Collections.emptyList(); userConfiguredPartitions = Collections.emptyList();
} }
if(userConfiguredPartitions.size() > 1) if(userConfiguredPartitions.size() > 1) {
throw DataXException.asDataXException(AdsWriterErrorCode.ODPS_PARTITION_FAILED, ""); throw DataXException.asDataXException(AdsWriterErrorCode.ODPS_PARTITION_FAILED, "");
}
if(userConfiguredPartitions.size() == 0) { if(userConfiguredPartitions.size() == 0) {
loadAdsData(adsHelper, odpsTableName,null); loadAdsData(adsHelper, odpsTableName,null);
...@@ -304,7 +307,7 @@ public class AdsWriter extends Writer { ...@@ -304,7 +307,7 @@ public class AdsWriter extends Writer {
private Configuration writerSliceConfig; private Configuration writerSliceConfig;
private OdpsWriter.Task odpsWriterTaskProxy = new OdpsWriter.Task(); private OdpsWriter.Task odpsWriterTaskProxy = new OdpsWriter.Task();
private String writeMode; private String writeMode;
private String schema; private String schema;
private String table; private String table;
...@@ -312,17 +315,28 @@ public class AdsWriter extends Writer { ...@@ -312,17 +315,28 @@ public class AdsWriter extends Writer {
// warn: 只有在insert, stream模式才有, 对于load模式表明为odps临时表了 // warn: 只有在insert, stream模式才有, 对于load模式表明为odps临时表了
private TableInfo tableInfo; private TableInfo tableInfo;
private String writeProxy;
AdsProxy proxy = null;
@Override @Override
public void init() { public void init() {
writerSliceConfig = super.getPluginJobConf(); writerSliceConfig = super.getPluginJobConf();
this.writeMode = this.writerSliceConfig.getString(Key.WRITE_MODE); this.writeMode = this.writerSliceConfig.getString(Key.WRITE_MODE);
this.schema = writerSliceConfig.getString(Key.SCHEMA); this.schema = writerSliceConfig.getString(Key.SCHEMA);
this.table = writerSliceConfig.getString(Key.ADS_TABLE); this.table = writerSliceConfig.getString(Key.ADS_TABLE);
if(Constant.LOADMODE.equalsIgnoreCase(this.writeMode)) { if(Constant.LOADMODE.equalsIgnoreCase(this.writeMode)) {
odpsWriterTaskProxy.setPluginJobConf(writerSliceConfig); odpsWriterTaskProxy.setPluginJobConf(writerSliceConfig);
odpsWriterTaskProxy.init(); odpsWriterTaskProxy.init();
} else if(Constant.INSERTMODE.equalsIgnoreCase(this.writeMode) || Constant.STREAMMODE.equalsIgnoreCase(this.writeMode)) { } else if(Constant.INSERTMODE.equalsIgnoreCase(this.writeMode) || Constant.STREAMMODE.equalsIgnoreCase(this.writeMode)) {
if (Constant.STREAMMODE.equalsIgnoreCase(this.writeMode)) {
this.writeProxy = "datax";
} else {
this.writeProxy = this.writerSliceConfig.getString("writeProxy", "adbClient");
}
this.writerSliceConfig.set("writeProxy", this.writeProxy);
try { try {
this.tableInfo = AdsUtil.createAdsHelper(this.writerSliceConfig).getTableInfo(this.table); this.tableInfo = AdsUtil.createAdsHelper(this.writerSliceConfig).getTableInfo(this.table);
} catch (AdsException e) { } catch (AdsException e) {
...@@ -361,7 +375,12 @@ public class AdsWriter extends Writer { ...@@ -361,7 +375,12 @@ public class AdsWriter extends Writer {
List<String> columns = writerSliceConfig.getList(Key.COLUMN, String.class); List<String> columns = writerSliceConfig.getList(Key.COLUMN, String.class);
Connection connection = AdsUtil.getAdsConnect(this.writerSliceConfig); Connection connection = AdsUtil.getAdsConnect(this.writerSliceConfig);
TaskPluginCollector taskPluginCollector = super.getTaskPluginCollector(); TaskPluginCollector taskPluginCollector = super.getTaskPluginCollector();
AdsInsertProxy proxy = new AdsInsertProxy(schema + "." + table, columns, writerSliceConfig, taskPluginCollector, this.tableInfo);
if (StringUtils.equalsIgnoreCase(this.writeProxy, "adbClient")) {
this.proxy = new AdsClientProxy(table, columns, writerSliceConfig, taskPluginCollector, this.tableInfo);
} else {
this.proxy = new AdsInsertProxy(schema + "." + table, columns, writerSliceConfig, taskPluginCollector, this.tableInfo);
}
proxy.startWriteWithConnection(recordReceiver, connection, columnNumber); proxy.startWriteWithConnection(recordReceiver, connection, columnNumber);
} }
} }
...@@ -381,6 +400,9 @@ public class AdsWriter extends Writer { ...@@ -381,6 +400,9 @@ public class AdsWriter extends Writer {
odpsWriterTaskProxy.destroy(); odpsWriterTaskProxy.destroy();
} else { } else {
//do noting until now //do noting until now
if (null != this.proxy) {
this.proxy.closeResource();
}
} }
} }
} }
......
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/insert/AdsClientProxy.java 0 → 100644
View file @ a57425d5
package com.alibaba.datax.plugin.writer.adswriter.insert;
import com.alibaba.cloud.analyticdb.adbclient.AdbClient;
import com.alibaba.cloud.analyticdb.adbclient.AdbClientException;
import com.alibaba.cloud.analyticdb.adbclient.DatabaseConfig;
import com.alibaba.cloud.analyticdb.adbclient.Row;
import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
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.core.transport.record.DefaultRecord;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
import com.alibaba.datax.plugin.writer.adswriter.AdsWriterErrorCode;
import com.alibaba.datax.plugin.writer.adswriter.ads.TableInfo;
import com.alibaba.datax.plugin.writer.adswriter.util.Constant;
import com.alibaba.datax.plugin.writer.adswriter.util.Key;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.Types;
import java.util.*;
public class AdsClientProxy implements AdsProxy {
private static final Logger LOG = LoggerFactory.getLogger(AdsClientProxy.class);
private String table;
private TaskPluginCollector taskPluginCollector;
public Configuration configuration;
// columnName: <java sql type, ads type name>
private Map<String, Pair<Integer, String>> adsTableColumnsMetaData;
private Map<String, Pair<Integer, String>> userConfigColumnsMetaData;
private boolean useRawData[];
private AdbClient adbClient;
/**
* warn: not support columns as *
*/
public AdsClientProxy(String table, List<String> columns, Configuration configuration,
TaskPluginCollector taskPluginCollector, TableInfo tableInfo) {
this.configuration = configuration;
this.taskPluginCollector = taskPluginCollector;
this.adsTableColumnsMetaData = AdsInsertUtil.getColumnMetaData(tableInfo, columns);
this.userConfigColumnsMetaData = new HashMap<String, Pair<Integer, String>>();
List<String> adsColumnsNames = tableInfo.getColumnsNames();
// 要使用用户配置的column顺序
this.useRawData = new boolean[columns.size()];
for (int i = 0; i < columns.size(); i++) {
String oriEachColumn = columns.get(i);
String eachColumn = oriEachColumn;
// 防御性保留字
if (eachColumn.startsWith(Constant.ADS_QUOTE_CHARACTER)
&& eachColumn.endsWith(Constant.ADS_QUOTE_CHARACTER)) {
eachColumn = eachColumn.substring(1, eachColumn.length() - 1);
}
for (String eachAdsColumn : adsColumnsNames) {
if (eachColumn.equalsIgnoreCase(eachAdsColumn)) {
Pair<Integer, String> eachMeta = this.adsTableColumnsMetaData.get(eachAdsColumn);
this.userConfigColumnsMetaData.put(oriEachColumn, eachMeta);
int columnSqltype = eachMeta.getLeft();
switch (columnSqltype) {
case Types.DATE:
case Types.TIME:
case Types.TIMESTAMP:
this.useRawData[i] = false;
break;
default:
this.useRawData[i] = true;
break;
}
}
}
}
DatabaseConfig databaseConfig = new DatabaseConfig();
String url = configuration.getString(Key.ADS_URL);
String[] hostAndPort = StringUtils.split(url, ":");
if (hostAndPort.length != 2) {
throw DataXException.asDataXException(AdsWriterErrorCode.INVALID_CONFIG_VALUE,
"url should be in host:port format!");
}
databaseConfig.setHost(hostAndPort[0]);
databaseConfig.setPort(Integer.parseInt(hostAndPort[1]));
databaseConfig.setUser(configuration.getString(Key.USERNAME));
databaseConfig.setPassword(configuration.getString(Key.PASSWORD));
databaseConfig.setDatabase(configuration.getString(Key.SCHEMA));
databaseConfig.setTable(Collections.singletonList(table));
this.table = table;
databaseConfig.setColumns(table, columns);
// 如果出现insert失败，是否跳过
boolean ignoreInsertError = configuration.getBool("ignoreInsertError", false);
databaseConfig.setIgnoreInsertError(ignoreInsertError);
// If the value of column is empty, set null
boolean emptyAsNull = configuration.getBool(Key.EMPTY_AS_NULL, false);
databaseConfig.setEmptyAsNull(emptyAsNull);
// 使用insert ignore into方式进行插入
boolean ignoreInsert = configuration.getBool(Key.IGNORE_INSERT, false);
databaseConfig.setInsertIgnore(ignoreInsert);
// commit时，写入ADB出现异常时重试的3次
int retryTimes = configuration.getInt(Key.RETRY_CONNECTION_TIME, Constant.DEFAULT_RETRY_TIMES);
databaseConfig.setRetryTimes(retryTimes);
// 重试间隔的时间为1s，单位是ms
int retryIntervalTime = configuration.getInt(Key.RETRY_INTERVAL_TIME, 1000);
databaseConfig.setRetryIntervalTime(retryIntervalTime);
// 设置自动提交的SQL长度（单位Byte），默认为32KB，一般不建议设置
int commitSize = configuration.getInt("commitSize", 32768);
databaseConfig.setCommitSize(commitSize);
// sdk默认为true
boolean partitionBatch = configuration.getBool("partitionBatch", true);
databaseConfig.setPartitionBatch(partitionBatch);
// 设置写入adb时的并发线程数，默认4，针对配置的所有表
int parallelNumber = configuration.getInt("parallelNumber", 4);
databaseConfig.setParallelNumber(parallelNumber);
// 设置client中使用的logger对象，此处使用slf4j.Logger
databaseConfig.setLogger(AdsClientProxy.LOG);
// 设置在拼接insert sql时是否需要带上字段名，默认为true
boolean insertWithColumnName = configuration.getBool("insertWithColumnName", true);
databaseConfig.setInsertWithColumnName(insertWithColumnName);
// sdk 默认值为true
boolean shareDataSource = configuration.getBool("shareDataSource", true);
databaseConfig.setShareDataSource(shareDataSource);
String password = databaseConfig.getPassword();
databaseConfig.setPassword(password.replaceAll(".", "*"));
// 避免敏感信息直接打印
LOG.info("Adb database config is : {}", JSON.toJSONString(databaseConfig));
databaseConfig.setPassword(password);
// Initialize AdbClient，初始化实例之后，databaseConfig的配置信息不能再修改
this.adbClient = new AdbClient(databaseConfig);
}
@Override
public void startWriteWithConnection(RecordReceiver recordReceiver, Connection connection, int columnNumber) {
try {
Record record;
while ((record = recordReceiver.getFromReader()) != null) {
Row row = new Row();
List<Object> values = new ArrayList<Object>();
this.prepareColumnTypeValue(record, values);
row.setColumnValues(values);
try {
this.adbClient.addRow(this.table, row);
} catch (AdbClientException e) {
if (101 == e.getCode()) {
for (String each : e.getErrData()) {
Record dirtyData = new DefaultRecord();
dirtyData.addColumn(new StringColumn(each));
this.taskPluginCollector.collectDirtyRecord(dirtyData, e.getMessage());
}
} else {
throw e;
}
}
}
try {
this.adbClient.commit();
} catch (AdbClientException e) {
if (101 == e.getCode()) {
for (String each : e.getErrData()) {
Record dirtyData = new DefaultRecord();
dirtyData.addColumn(new StringColumn(each));
this.taskPluginCollector.collectDirtyRecord(dirtyData, e.getMessage());
}
} else {
throw e;
}
}
} catch (Exception e) {
throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
} finally {
DBUtil.closeDBResources(null, null, connection);
}
}
private void prepareColumnTypeValue(Record record, List<Object> values) {
for (int i = 0; i < this.useRawData.length; i++) {
Column column = record.getColumn(i);
if (this.useRawData[i]) {
values.add(column.getRawData());
} else {
values.add(column.asString());
}
}
}
@Override
public void closeResource() {
try {
LOG.info("stop the adbClient");
this.adbClient.stop();
} catch (Exception e) {
LOG.warn("stop adbClient meet a exception, ignore it: {}", e.getMessage(), e);
}
}
}
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/insert/AdsInsertProxy.java
View file @ a57425d5
...@@ -34,7 +34,7 @@ import java.util.zip.CRC32; ...@@ -34,7 +34,7 @@ import java.util.zip.CRC32;
import java.util.zip.Checksum; import java.util.zip.Checksum;
public class AdsInsertProxy { public class AdsInsertProxy implements AdsProxy {
private static final Logger LOG = LoggerFactory private static final Logger LOG = LoggerFactory
.getLogger(AdsInsertProxy.class); .getLogger(AdsInsertProxy.class);
...@@ -46,9 +46,9 @@ public class AdsInsertProxy { ...@@ -46,9 +46,9 @@ public class AdsInsertProxy {
private TaskPluginCollector taskPluginCollector; private TaskPluginCollector taskPluginCollector;
private Configuration configuration; private Configuration configuration;
private Boolean emptyAsNull; private Boolean emptyAsNull;
private String writeMode; private String writeMode;
private String insertSqlPrefix; private String insertSqlPrefix;
private String deleteSqlPrefix; private String deleteSqlPrefix;
private int opColumnIndex; private int opColumnIndex;
...@@ -58,10 +58,10 @@ public class AdsInsertProxy { ...@@ -58,10 +58,10 @@ public class AdsInsertProxy {
private Map<String, Pair<Integer, String>> userConfigColumnsMetaData; private Map<String, Pair<Integer, String>> userConfigColumnsMetaData;
// columnName: index @ ads column // columnName: index @ ads column
private Map<String, Integer> primaryKeyNameIndexMap; private Map<String, Integer> primaryKeyNameIndexMap;
private int retryTimeUpperLimit; private int retryTimeUpperLimit;
private Connection currentConnection; private Connection currentConnection;
private String partitionColumn; private String partitionColumn;
private int partitionColumnIndex = -1; private int partitionColumnIndex = -1;
private int partitionCount; private int partitionCount;
...@@ -80,14 +80,14 @@ public class AdsInsertProxy { ...@@ -80,14 +80,14 @@ public class AdsInsertProxy {
Key.RETRY_CONNECTION_TIME, Constant.DEFAULT_RETRY_TIMES); Key.RETRY_CONNECTION_TIME, Constant.DEFAULT_RETRY_TIMES);
this.partitionCount = tableInfo.getPartitionCount(); this.partitionCount = tableInfo.getPartitionCount();
this.partitionColumn = tableInfo.getPartitionColumn(); this.partitionColumn = tableInfo.getPartitionColumn();
//目前ads新建的表如果未插入数据不能通过select colums from table where 1=2，获取列信息，需要读取ads数据字典 //目前ads新建的表如果未插入数据不能通过select colums from table where 1=2，获取列信息，需要读取ads数据字典
//not this: this.resultSetMetaData = DBUtil.getColumnMetaData(connection, this.table, StringUtils.join(this.columns, ",")); //not this: this.resultSetMetaData = DBUtil.getColumnMetaData(connection, this.table, StringUtils.join(this.columns, ","));
//no retry here(fetch meta data) 注意实时表列换序的可能 //no retry here(fetch meta data) 注意实时表列换序的可能
this.adsTableColumnsMetaData = AdsInsertUtil.getColumnMetaData(tableInfo, this.columns); this.adsTableColumnsMetaData = AdsInsertUtil.getColumnMetaData(tableInfo, this.columns);
this.userConfigColumnsMetaData = new HashMap<String, Pair<Integer, String>>(); this.userConfigColumnsMetaData = new HashMap<String, Pair<Integer, String>>();
List<String> primaryKeyColumnName = tableInfo.getPrimaryKeyColumns(); List<String> primaryKeyColumnName = tableInfo.getPrimaryKeyColumns();
List<String> adsColumnsNames = tableInfo.getColumnsNames(); List<String> adsColumnsNames = tableInfo.getColumnsNames();
this.primaryKeyNameIndexMap = new HashMap<String, Integer>(); this.primaryKeyNameIndexMap = new HashMap<String, Integer>();
//warn: 要使用用户配置的column顺序, 不要使用从ads元数据获取的column顺序, 原来复用load列顺序其实有问题的 //warn: 要使用用户配置的column顺序, 不要使用从ads元数据获取的column顺序, 原来复用load列顺序其实有问题的
...@@ -108,7 +108,7 @@ public class AdsInsertProxy { ...@@ -108,7 +108,7 @@ public class AdsInsertProxy {
this.userConfigColumnsMetaData.put(oriEachColumn, this.adsTableColumnsMetaData.get(eachAdsColumn)); this.userConfigColumnsMetaData.put(oriEachColumn, this.adsTableColumnsMetaData.get(eachAdsColumn));
} }
} }
// 根据第几个column分区列排序，ads实时表只有一级分区、最多256个分区 // 根据第几个column分区列排序，ads实时表只有一级分区、最多256个分区
if (eachColumn.equalsIgnoreCase(this.partitionColumn)) { if (eachColumn.equalsIgnoreCase(this.partitionColumn)) {
this.partitionColumnIndex = i; this.partitionColumnIndex = i;
...@@ -117,8 +117,8 @@ public class AdsInsertProxy { ...@@ -117,8 +117,8 @@ public class AdsInsertProxy {
} }
public void startWriteWithConnection(RecordReceiver recordReceiver, public void startWriteWithConnection(RecordReceiver recordReceiver,
Connection connection, Connection connection,
int columnNumber) { int columnNumber) {
this.currentConnection = connection; this.currentConnection = connection;
int batchSize = this.configuration.getInt(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE); int batchSize = this.configuration.getInt(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
// 默认情况下bufferSize需要和batchSize一致 // 默认情况下bufferSize需要和batchSize一致
...@@ -164,13 +164,13 @@ public class AdsInsertProxy { ...@@ -164,13 +164,13 @@ public class AdsInsertProxy {
OperationType operationType = OperationType.asOperationType(optionColumnValue); OperationType operationType = OperationType.asOperationType(optionColumnValue);
if (operationType.isInsertTemplate()) { if (operationType.isInsertTemplate()) {
writeBuffer.add(record); writeBuffer.add(record);
if (this.lastDmlMode == null || this.lastDmlMode == Constant.INSERTMODE ) { if (this.lastDmlMode == null || this.lastDmlMode == Constant.INSERTMODE) {
this.lastDmlMode = Constant.INSERTMODE; this.lastDmlMode = Constant.INSERTMODE;
if (writeBuffer.size() >= bufferSize) { if (writeBuffer.size() >= bufferSize) {
this.doBatchRecordWithPartitionSort(writeBuffer, Constant.INSERTMODE, bufferSize, batchSize); this.doBatchRecordWithPartitionSort(writeBuffer, Constant.INSERTMODE, bufferSize, batchSize);
writeBuffer.clear(); writeBuffer.clear();
} }
} else { } else {
this.lastDmlMode = Constant.INSERTMODE; this.lastDmlMode = Constant.INSERTMODE;
// 模式变换触发一次提交ads delete, 并进入insert模式 // 模式变换触发一次提交ads delete, 并进入insert模式
this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize); this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize);
...@@ -178,7 +178,7 @@ public class AdsInsertProxy { ...@@ -178,7 +178,7 @@ public class AdsInsertProxy {
} }
} else if (operationType.isDeleteTemplate()) { } else if (operationType.isDeleteTemplate()) {
deleteBuffer.add(record); deleteBuffer.add(record);
if (this.lastDmlMode == null || this.lastDmlMode == Constant.DELETEMODE ) { if (this.lastDmlMode == null || this.lastDmlMode == Constant.DELETEMODE) {
this.lastDmlMode = Constant.DELETEMODE; this.lastDmlMode = Constant.DELETEMODE;
if (deleteBuffer.size() >= bufferSize) { if (deleteBuffer.size() >= bufferSize) {
this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize); this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize);
...@@ -196,14 +196,14 @@ public class AdsInsertProxy { ...@@ -196,14 +196,14 @@ public class AdsInsertProxy {
} }
} }
} }
if (!writeBuffer.isEmpty()) { if (!writeBuffer.isEmpty()) {
//doOneRecord(writeBuffer, Constant.INSERTMODE); //doOneRecord(writeBuffer, Constant.INSERTMODE);
this.doBatchRecordWithPartitionSort(writeBuffer, Constant.INSERTMODE, bufferSize, batchSize); this.doBatchRecordWithPartitionSort(writeBuffer, Constant.INSERTMODE, bufferSize, batchSize);
writeBuffer.clear(); writeBuffer.clear();
} }
// 2个缓冲最多一个不为空同时 // 2个缓冲最多一个不为空同时
if (null!= deleteBuffer && !deleteBuffer.isEmpty()) { if (null != deleteBuffer && !deleteBuffer.isEmpty()) {
//doOneRecord(deleteBuffer, Constant.DELETEMODE); //doOneRecord(deleteBuffer, Constant.DELETEMODE);
this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize); this.doBatchRecordWithPartitionSort(deleteBuffer, Constant.DELETEMODE, bufferSize, batchSize);
deleteBuffer.clear(); deleteBuffer.clear();
...@@ -216,14 +216,14 @@ public class AdsInsertProxy { ...@@ -216,14 +216,14 @@ public class AdsInsertProxy {
DBUtil.closeDBResources(null, null, connection); DBUtil.closeDBResources(null, null, connection);
} }
} }
/** /**
* @param bufferSize datax缓冲记录条数 * @param bufferSize datax缓冲记录条数
* @param batchSize datax向ads系统一次发送数据条数 * @param batchSize datax向ads系统一次发送数据条数
* @param buffer datax缓冲区 * @param buffer datax缓冲区
* @param mode 实时表模式insert 或者 stream * @param mode 实时表模式insert 或者 stream
* */ */
private void doBatchRecordWithPartitionSort(List<Record> buffer, String mode, int bufferSize, int batchSize) throws SQLException{ private void doBatchRecordWithPartitionSort(List<Record> buffer, String mode, int bufferSize, int batchSize) throws SQLException {
//warn: 排序会影响数据插入顺序, 如果源头没有数据约束, 排序可能造成数据不一致, 快速排序是一种不稳定的排序算法 //warn: 排序会影响数据插入顺序, 如果源头没有数据约束, 排序可能造成数据不一致, 快速排序是一种不稳定的排序算法
//warn: 不明确配置bufferSize或者小于batchSize的情况下，不要进行排序;如果缓冲区实际内容条数少于batchSize也不排序了，最后一次的余量 //warn: 不明确配置bufferSize或者小于batchSize的情况下，不要进行排序;如果缓冲区实际内容条数少于batchSize也不排序了，最后一次的余量
int recordBufferedNumber = buffer.size(); int recordBufferedNumber = buffer.size();
...@@ -261,7 +261,7 @@ public class AdsInsertProxy { ...@@ -261,7 +261,7 @@ public class AdsInsertProxy {
return true; return true;
} }
}, this.retryTimeUpperLimit, 2000L, true, retryExceptionClasss); }, this.retryTimeUpperLimit, 2000L, true, retryExceptionClasss);
}catch (SQLException e) { } catch (SQLException e) {
LOG.warn(String.format("after retry %s times, doBatchRecord meet a exception: ", this.retryTimeUpperLimit), e); LOG.warn(String.format("after retry %s times, doBatchRecord meet a exception: ", this.retryTimeUpperLimit), e);
LOG.info("try to re execute for each record..."); LOG.info("try to re execute for each record...");
doOneRecord(buffer, mode); doOneRecord(buffer, mode);
...@@ -274,7 +274,7 @@ public class AdsInsertProxy { ...@@ -274,7 +274,7 @@ public class AdsInsertProxy {
DBUtilErrorCode.WRITE_DATA_ERROR, e); DBUtilErrorCode.WRITE_DATA_ERROR, e);
} }
} }
//warn: ADS 无法支持事物roll back都是不管用 //warn: ADS 无法支持事物roll back都是不管用
@SuppressWarnings("resource") @SuppressWarnings("resource")
private void doBatchRecordDml(List<Record> buffer, String mode) throws Exception { private void doBatchRecordDml(List<Record> buffer, String mode) throws Exception {
...@@ -313,7 +313,7 @@ public class AdsInsertProxy { ...@@ -313,7 +313,7 @@ public class AdsInsertProxy {
} else { } else {
try { try {
Throwable causeThrowable = eachException.getCause(); Throwable causeThrowable = eachException.getCause();
eachException = causeThrowable == null ? null : (Exception)causeThrowable; eachException = causeThrowable == null ? null : (Exception) causeThrowable;
} catch (Exception castException) { } catch (Exception castException) {
LOG.warn("doBatchRecordDml meet a no! retry exception: " + e.getMessage()); LOG.warn("doBatchRecordDml meet a no! retry exception: " + e.getMessage());
throw e; throw e;
...@@ -330,7 +330,7 @@ public class AdsInsertProxy { ...@@ -330,7 +330,7 @@ public class AdsInsertProxy {
DBUtil.closeDBResources(statement, null); DBUtil.closeDBResources(statement, null);
} }
} }
private void doOneRecord(List<Record> buffer, final String mode) { private void doOneRecord(List<Record> buffer, final String mode) {
List<Class<?>> retryExceptionClasss = new ArrayList<Class<?>>(); List<Class<?>> retryExceptionClasss = new ArrayList<Class<?>>();
retryExceptionClasss.add(com.mysql.jdbc.exceptions.jdbc4.CommunicationsException.class); retryExceptionClasss.add(com.mysql.jdbc.exceptions.jdbc4.CommunicationsException.class);
...@@ -350,7 +350,7 @@ public class AdsInsertProxy { ...@@ -350,7 +350,7 @@ public class AdsInsertProxy {
} }
} }
} }
@SuppressWarnings("resource") @SuppressWarnings("resource")
private void doOneRecordDml(Record record, String mode) throws Exception { private void doOneRecordDml(Record record, String mode) throws Exception {
Statement statement = null; Statement statement = null;
...@@ -380,7 +380,7 @@ public class AdsInsertProxy { ...@@ -380,7 +380,7 @@ public class AdsInsertProxy {
} else { } else {
try { try {
Throwable causeThrowable = eachException.getCause(); Throwable causeThrowable = eachException.getCause();
eachException = causeThrowable == null ? null : (Exception)causeThrowable; eachException = causeThrowable == null ? null : (Exception) causeThrowable;
} catch (Exception castException) { } catch (Exception castException) {
LOG.warn("doOneDml meet a no! retry exception: " + e.getMessage()); LOG.warn("doOneDml meet a no! retry exception: " + e.getMessage());
throw e; throw e;
...@@ -397,7 +397,7 @@ public class AdsInsertProxy { ...@@ -397,7 +397,7 @@ public class AdsInsertProxy {
DBUtil.closeDBResources(statement, null); DBUtil.closeDBResources(statement, null);
} }
} }
private boolean isRetryable(Throwable e) { private boolean isRetryable(Throwable e) {
Class<?> meetExceptionClass = e.getClass(); Class<?> meetExceptionClass = e.getClass();
if (meetExceptionClass == com.mysql.jdbc.exceptions.jdbc4.CommunicationsException.class) { if (meetExceptionClass == com.mysql.jdbc.exceptions.jdbc4.CommunicationsException.class) {
...@@ -408,7 +408,7 @@ public class AdsInsertProxy { ...@@ -408,7 +408,7 @@ public class AdsInsertProxy {
} }
return false; return false;
} }
private String generateDmlSql(Connection connection, Record record, String mode) throws SQLException { private String generateDmlSql(Connection connection, Record record, String mode) throws SQLException {
String sql = null; String sql = null;
StringBuilder sqlSb = new StringBuilder(); StringBuilder sqlSb = new StringBuilder();
...@@ -417,7 +417,7 @@ public class AdsInsertProxy { ...@@ -417,7 +417,7 @@ public class AdsInsertProxy {
sqlSb.append("("); sqlSb.append("(");
int columnsSize = this.columns.size(); int columnsSize = this.columns.size();
for (int i = 0; i < columnsSize; i++) { for (int i = 0; i < columnsSize; i++) {
if((i + 1) != columnsSize) { if ((i + 1) != columnsSize) {
sqlSb.append("?,"); sqlSb.append("?,");
} else { } else {
sqlSb.append("?"); sqlSb.append("?");
...@@ -431,7 +431,7 @@ public class AdsInsertProxy { ...@@ -431,7 +431,7 @@ public class AdsInsertProxy {
if (Constant.STREAMMODE.equalsIgnoreCase(this.writeMode)) { if (Constant.STREAMMODE.equalsIgnoreCase(this.writeMode)) {
if (preparedParamsIndex >= this.opColumnIndex) { if (preparedParamsIndex >= this.opColumnIndex) {
preparedParamsIndex = i + 1; preparedParamsIndex = i + 1;
} }
} }
String columnName = this.columns.get(i); String columnName = this.columns.get(i);
int columnSqltype = this.userConfigColumnsMetaData.get(columnName).getLeft(); int columnSqltype = this.userConfigColumnsMetaData.get(columnName).getLeft();
...@@ -446,7 +446,7 @@ public class AdsInsertProxy { ...@@ -446,7 +446,7 @@ public class AdsInsertProxy {
int entrySetSize = primaryEntrySet.size(); int entrySetSize = primaryEntrySet.size();
int i = 0; int i = 0;
for (Entry<String, Integer> eachEntry : primaryEntrySet) { for (Entry<String, Integer> eachEntry : primaryEntrySet) {
if((i + 1) != entrySetSize) { if ((i + 1) != entrySetSize) {
sqlSb.append(String.format(" (%s = ?) and ", eachEntry.getKey())); sqlSb.append(String.format(" (%s = ?) and ", eachEntry.getKey()));
} else { } else {
sqlSb.append(String.format(" (%s = ?) ", eachEntry.getKey())); sqlSb.append(String.format(" (%s = ?) ", eachEntry.getKey()));
...@@ -463,7 +463,7 @@ public class AdsInsertProxy { ...@@ -463,7 +463,7 @@ public class AdsInsertProxy {
int columnSqlType = this.userConfigColumnsMetaData.get(columnName).getLeft(); int columnSqlType = this.userConfigColumnsMetaData.get(columnName).getLeft();
int primaryKeyInUserConfigIndex = this.primaryKeyNameIndexMap.get(columnName); int primaryKeyInUserConfigIndex = this.primaryKeyNameIndexMap.get(columnName);
if (primaryKeyInUserConfigIndex >= this.opColumnIndex) { if (primaryKeyInUserConfigIndex >= this.opColumnIndex) {
primaryKeyInUserConfigIndex ++; primaryKeyInUserConfigIndex++;
} }
prepareColumnTypeValue(statement, columnSqlType, record.getColumn(primaryKeyInUserConfigIndex), i, columnName); prepareColumnTypeValue(statement, columnSqlType, record.getColumn(primaryKeyInUserConfigIndex), i, columnName);
i++; i++;
...@@ -473,8 +473,8 @@ public class AdsInsertProxy { ...@@ -473,8 +473,8 @@ public class AdsInsertProxy {
} }
return sql; return sql;
} }
private void appendDmlSqlValues(Connection connection, Record record, StringBuilder sqlSb, String mode) throws SQLException { private void appendDmlSqlValues(Connection connection, Record record, StringBuilder sqlSb, String mode) throws SQLException {
String sqlResult = this.generateDmlSql(connection, record, mode); String sqlResult = this.generateDmlSql(connection, record, mode);
if (mode.equalsIgnoreCase(Constant.INSERTMODE)) { if (mode.equalsIgnoreCase(Constant.INSERTMODE)) {
sqlSb.append(","); sqlSb.append(",");
...@@ -508,21 +508,21 @@ public class AdsInsertProxy { ...@@ -508,21 +508,21 @@ public class AdsInsertProxy {
case Types.DECIMAL: case Types.DECIMAL:
case Types.REAL: case Types.REAL:
String numValue = column.asString(); String numValue = column.asString();
if(emptyAsNull && "".equals(numValue) || numValue == null){ if (emptyAsNull && "".equals(numValue) || numValue == null) {
//statement.setObject(preparedPatamIndex + 1, null); //statement.setObject(preparedPatamIndex + 1, null);
statement.setNull(preparedPatamIndex + 1, Types.BIGINT); statement.setNull(preparedPatamIndex + 1, Types.BIGINT);
} else{ } else {
statement.setLong(preparedPatamIndex + 1, column.asLong()); statement.setLong(preparedPatamIndex + 1, column.asLong());
} }
break; break;
case Types.FLOAT: case Types.FLOAT:
case Types.DOUBLE: case Types.DOUBLE:
String floatValue = column.asString(); String floatValue = column.asString();
if(emptyAsNull && "".equals(floatValue) || floatValue == null){ if (emptyAsNull && "".equals(floatValue) || floatValue == null) {
//statement.setObject(preparedPatamIndex + 1, null); //statement.setObject(preparedPatamIndex + 1, null);
statement.setNull(preparedPatamIndex + 1, Types.DOUBLE); statement.setNull(preparedPatamIndex + 1, Types.DOUBLE);
} else{ } else {
statement.setDouble(preparedPatamIndex + 1, column.asDouble()); statement.setDouble(preparedPatamIndex + 1, column.asDouble());
} }
break; break;
...@@ -535,13 +535,13 @@ public class AdsInsertProxy { ...@@ -535,13 +535,13 @@ public class AdsInsertProxy {
} else { } else {
statement.setLong(preparedPatamIndex + 1, longValue); statement.setLong(preparedPatamIndex + 1, longValue);
} }
break; break;
case Types.DATE: case Types.DATE:
java.sql.Date sqlDate = null; java.sql.Date sqlDate = null;
try { try {
if("".equals(column.getRawData())) { if ("".equals(column.getRawData())) {
utilDate = null; utilDate = null;
} else { } else {
utilDate = column.asDate(); utilDate = column.asDate();
...@@ -550,17 +550,17 @@ public class AdsInsertProxy { ...@@ -550,17 +550,17 @@ public class AdsInsertProxy {
throw new SQLException(String.format( throw new SQLException(String.format(
"Date 类型转换错误：[%s]", column)); "Date 类型转换错误：[%s]", column));
} }
if (null != utilDate) { if (null != utilDate) {
sqlDate = new java.sql.Date(utilDate.getTime()); sqlDate = new java.sql.Date(utilDate.getTime());
} }
statement.setDate(preparedPatamIndex + 1, sqlDate); statement.setDate(preparedPatamIndex + 1, sqlDate);
break; break;
case Types.TIME: case Types.TIME:
java.sql.Time sqlTime = null; java.sql.Time sqlTime = null;
try { try {
if("".equals(column.getRawData())) { if ("".equals(column.getRawData())) {
utilDate = null; utilDate = null;
} else { } else {
utilDate = column.asDate(); utilDate = column.asDate();
...@@ -579,7 +579,7 @@ public class AdsInsertProxy { ...@@ -579,7 +579,7 @@ public class AdsInsertProxy {
case Types.TIMESTAMP: case Types.TIMESTAMP:
java.sql.Timestamp sqlTimestamp = null; java.sql.Timestamp sqlTimestamp = null;
try { try {
if("".equals(column.getRawData())) { if ("".equals(column.getRawData())) {
utilDate = null; utilDate = null;
} else { } else {
utilDate = column.asDate(); utilDate = column.asDate();
...@@ -597,14 +597,14 @@ public class AdsInsertProxy { ...@@ -597,14 +597,14 @@ public class AdsInsertProxy {
break; break;
case Types.BOOLEAN: case Types.BOOLEAN:
//case Types.BIT: ads 没有bit //case Types.BIT: ads 没有bit
Boolean booleanValue = column.asBoolean(); Boolean booleanValue = column.asBoolean();
if (null == booleanValue) { if (null == booleanValue) {
statement.setNull(preparedPatamIndex + 1, Types.BOOLEAN); statement.setNull(preparedPatamIndex + 1, Types.BOOLEAN);
} else { } else {
statement.setBoolean(preparedPatamIndex + 1, booleanValue); statement.setBoolean(preparedPatamIndex + 1, booleanValue);
} }
break; break;
default: default:
Pair<Integer, String> columnMetaPair = this.userConfigColumnsMetaData.get(columnName); Pair<Integer, String> columnMetaPair = this.userConfigColumnsMetaData.get(columnName);
...@@ -616,7 +616,7 @@ public class AdsInsertProxy { ...@@ -616,7 +616,7 @@ public class AdsInsertProxy {
columnName, columnMetaPair.getRight(), columnMetaPair.getLeft())); columnName, columnMetaPair.getRight(), columnMetaPair.getLeft()));
} }
} }
private static int getHashPartition(String value, int totalHashPartitionNum) { private static int getHashPartition(String value, int totalHashPartitionNum) {
long crc32 = (value == null ? getCRC32("-1") : getCRC32(value)); long crc32 = (value == null ? getCRC32("-1") : getCRC32(value));
return (int) (crc32 % totalHashPartitionNum); return (int) (crc32 % totalHashPartitionNum);
...@@ -628,4 +628,8 @@ public class AdsInsertProxy { ...@@ -628,4 +628,8 @@ public class AdsInsertProxy {
checksum.update(bytes, 0, bytes.length); checksum.update(bytes, 0, bytes.length);
return checksum.getValue(); return checksum.getValue();
} }
@Override
public void closeResource() {
}
} }
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/insert/AdsProxy.java 0 → 100644
View file @ a57425d5
package com.alibaba.datax.plugin.writer.adswriter.insert;
import com.alibaba.datax.common.plugin.RecordReceiver;
import java.sql.Connection;
public interface AdsProxy {
public abstract void startWriteWithConnection(RecordReceiver recordReceiver, Connection connection,
int columnNumber);
public void closeResource();
}
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/load/AdsHelper.java
View file @ a57425d5
...@@ -121,7 +121,7 @@ public class AdsHelper { ...@@ -121,7 +121,7 @@ public class AdsHelper {
connection = DriverManager.getConnection(url, connectionProps); connection = DriverManager.getConnection(url, connectionProps);
statement = connection.createStatement(); statement = connection.createStatement();
// ads 表名、schema名不区分大小写, 提高用户易用性, 注意列顺序性 // ads 表名、schema名不区分大小写, 提高用户易用性, 注意列顺序性
String columnMetaSql = String.format("select ordinal_position,column_name,data_type,type_name,column_comment from information_schema.columns where lower(table_schema) = `'%s'` and lower(table_name) = `'%s'` order by ordinal_position", schema.toLowerCase(), table.toLowerCase()); String columnMetaSql = String.format("select ordinal_position,column_name,data_type,type_name,column_comment from information_schema.columns where table_schema = `'%s'` and table_name = `'%s'` order by ordinal_position", schema.toLowerCase(), table.toLowerCase());
LOG.info(String.format("检查列信息sql语句:%s", columnMetaSql)); LOG.info(String.format("检查列信息sql语句:%s", columnMetaSql));
rs = statement.executeQuery(columnMetaSql); rs = statement.executeQuery(columnMetaSql);
...@@ -145,7 +145,7 @@ public class AdsHelper { ...@@ -145,7 +145,7 @@ public class AdsHelper {
tableInfo.setTableName(table); tableInfo.setTableName(table);
DBUtil.closeDBResources(rs, statement, null); DBUtil.closeDBResources(rs, statement, null);
String tableMetaSql = String.format("select update_type, partition_type, partition_column, partition_count, primary_key_columns from information_schema.tables where lower(table_schema) = `'%s'` and lower(table_name) = `'%s'`", schema.toLowerCase(), table.toLowerCase()); String tableMetaSql = String.format("select update_type, partition_type, partition_column, partition_count, primary_key_columns from information_schema.tables where table_schema = `'%s'` and table_name = `'%s'`", schema.toLowerCase(), table.toLowerCase());
LOG.info(String.format("检查表信息sql语句:%s", tableMetaSql)); LOG.info(String.format("检查表信息sql语句:%s", tableMetaSql));
statement = connection.createStatement(); statement = connection.createStatement();
rs = statement.executeQuery(tableMetaSql); rs = statement.executeQuery(tableMetaSql);
......
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/util/Constant.java
View file @ a57425d5
...@@ -16,7 +16,7 @@ public class Constant { ...@@ -16,7 +16,7 @@ public class Constant {
public static final long DEFAULT_SOCKET_TIMEOUT = 3600000L; public static final long DEFAULT_SOCKET_TIMEOUT = 3600000L;
public static final int DEFAULT_RETRY_TIMES = 2; public static final int DEFAULT_RETRY_TIMES = 3;
public static final String INSERT_TEMPLATE = "insert into %s ( %s ) values "; public static final String INSERT_TEMPLATE = "insert into %s ( %s ) values ";
......
adswriter/src/main/java/com/alibaba/datax/plugin/writer/adswriter/util/Key.java
View file @ a57425d5
...@@ -29,7 +29,9 @@ public final class Key { ...@@ -29,7 +29,9 @@ public final class Key {
public final static String BATCH_SIZE = "batchSize"; public final static String BATCH_SIZE = "batchSize";
public final static String BUFFER_SIZE = "bufferSize"; public final static String BUFFER_SIZE = "bufferSize";
public final static String IGNORE_INSERT = "ignoreInsert";
public final static String PRE_SQL = "preSql"; public final static String PRE_SQL = "preSql";
public final static String POST_SQL = "postSql"; public final static String POST_SQL = "postSql";
...@@ -37,7 +39,9 @@ public final class Key { ...@@ -37,7 +39,9 @@ public final class Key {
public final static String SOCKET_TIMEOUT = "socketTimeout"; public final static String SOCKET_TIMEOUT = "socketTimeout";
public final static String RETRY_CONNECTION_TIME = "retryTimes"; public final static String RETRY_CONNECTION_TIME = "retryTimes";
public final static String RETRY_INTERVAL_TIME = "retryIntervalTime";
public final static String JDBC_URL_SUFFIX = "urlSuffix"; public final static String JDBC_URL_SUFFIX = "urlSuffix";
/** /**
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment