github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-02 17:22:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

bug fix and write speed optimization

Browse Source
This commit is contained in:
daizihao 2020-10-13 20:23:27 +08:00
parent 80860c224d
commit 00d8e9783d
2 changed files with 55 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Kudu11xHelper.java
View File

@ -111,7 +111,7 @@ public class Kudu11xHelper {
} catch (Exception e) { } catch (Exception e) {
throw DataXException.asDataXException(Kudu11xWriterErrorcode.GREATE_KUDU_TABLE_ERROR, e); throw DataXException.asDataXException(Kudu11xWriterErrorcode.GREATE_KUDU_TABLE_ERROR, e);
} finally { } finally {
AtomicInteger i = new AtomicInteger(5); AtomicInteger i = new AtomicInteger(10);
while (i.get() > 0) { while (i.get() > 0) {
try { try {
if (kuduClient.isCreateTableDone(tableName)) { if (kuduClient.isCreateTableDone(tableName)) {
@ -124,7 +124,7 @@ public class Kudu11xHelper {
} catch (KuduException e) { } catch (KuduException e) {
LOG.info("Wait for the table to be created..... " + i); LOG.info("Wait for the table to be created..... " + i);
try { try {
Thread.sleep(1000L); Thread.sleep(100L);
} catch (InterruptedException ex) { } catch (InterruptedException ex) {
ex.printStackTrace(); ex.printStackTrace();
} }

101
kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/KuduWriterTask.java
View File

@ -94,61 +94,66 @@ public class KuduWriterTask {
//增量更新 //增量更新
row = insert.getRow(); row = insert.getRow();
} }
List<Future<?>> futures = new ArrayList<>();
for (List<Configuration> columnList : columnLists) { for (List<Configuration> columnList : columnLists) {
Record finalRecord = record; Record finalRecord = record;
pool.submit(()->{ Future<?> future = pool.submit(() -> {
try {
for (Configuration col : columnList) { for (Configuration col : columnList) {
String name = col.getString(Key.NAME);
String name = col.getString(Key.NAME); ColumnType type = ColumnType.getByTypeName(col.getString(Key.TYPE, "string"));
ColumnType type = ColumnType.getByTypeName(col.getString(Key.TYPE, "string")); Column column = finalRecord.getColumn(col.getInt(Key.INDEX));
Column column = finalRecord.getColumn(col.getInt(Key.INDEX)); String rawData = column.asString();
String rawData = column.asString(); if (rawData == null) {
if (rawData == null) { synchronized (lock) {
synchronized (lock) { row.setNull(name);
row.setNull(name); }
continue;
}
switch (type) {
case INT:
synchronized (lock) {
row.addInt(name, Integer.parseInt(rawData));
}
break;
case LONG:
case BIGINT:
synchronized (lock) {
row.addLong(name, Long.parseLong(rawData));
}
break;
case FLOAT:
synchronized (lock) {
row.addFloat(name, Float.parseFloat(rawData));
}
break;
case DOUBLE:
synchronized (lock) {
row.addDouble(name, Double.parseDouble(rawData));
}
break;
case BOOLEAN:
synchronized (lock) {
row.addBoolean(name, Boolean.getBoolean(rawData));
}
break;
case STRING:
default:
synchronized (lock) {
row.addString(name, rawData);
}
} }
continue;
}
switch (type) {
case INT:
synchronized (lock) {
row.addInt(name, Integer.parseInt(rawData));
}
break;
case LONG:
case BIGINT:
synchronized (lock) {
row.addLong(name, Long.parseLong(rawData));
}
break;
case FLOAT:
synchronized (lock) {
row.addFloat(name, Float.parseFloat(rawData));
}
break;
case DOUBLE:
synchronized (lock) {
row.addDouble(name, Double.parseDouble(rawData));
}
break;
case BOOLEAN:
synchronized (lock) {
row.addBoolean(name, Boolean.getBoolean(rawData));
}
break;
case STRING:
default:
synchronized (lock) {
row.addString(name, rawData);
}
} }
} finally {
countDownLatch.countDown();
} }
countDownLatch.countDown();
}); });
futures.add(future);
} }
countDownLatch.await(); countDownLatch.await();
for (Future<?> future : futures) {
future.get();
}
try { try {
RetryUtil.executeWithRetry(() -> { RetryUtil.executeWithRetry(() -> {
if (isUpsert) { if (isUpsert) {
@ -173,7 +178,7 @@ public class KuduWriterTask {
LOG.warn("Since you have configured \"skipFail\" to be true, this record will be skipped !"); LOG.warn("Since you have configured \"skipFail\" to be true, this record will be skipped !");
taskPluginCollector.collectDirtyRecord(record, e.getMessage()); taskPluginCollector.collectDirtyRecord(record, e.getMessage());
} else { } else {
throw e; throw DataXException.asDataXException(Kudu11xWriterErrorcode.PUT_KUDU_ERROR, e.getMessage());
} }
} }
} }