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

reduce the getBytes operations

Browse Source
This commit is contained in:
hffariel 2021-09-23 17:15:14 +08:00
parent e7d879185b
commit fe43e20ea9
3 changed files with 12 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
starrockswriter/src/main/java/com/starrocks/connector/datax/plugin/writer/starrockswriter/manager/StarRocksFlushTuple.java
View File

@ -6,9 +6,9 @@ public class StarRocksFlushTuple {
private String label; private String label;
private Long bytes; private Long bytes;
private List<String> rows; private List<byte[]> rows;
public StarRocksFlushTuple(String label, Long bytes, List<String> rows) { public StarRocksFlushTuple(String label, Long bytes, List<byte[]> rows) {
this.label = label; this.label = label;
this.bytes = bytes; this.bytes = bytes;
this.rows = rows; this.rows = rows;
@ -16,5 +16,5 @@ public class StarRocksFlushTuple {
public String getLabel() { return label; } public String getLabel() { return label; }
public Long getBytes() { return bytes; } public Long getBytes() { return bytes; }
public List<String> getRows() { return rows; } public List<byte[]> getRows() { return rows; }
} }

10
starrockswriter/src/main/java/com/starrocks/connector/datax/plugin/writer/starrockswriter/manager/StarRocksStreamLoadVisitor.java
View File

@ -93,13 +93,13 @@ public class StarRocksStreamLoadVisitor {
} }
} }
private byte[] joinRows(List<String> rows, int totalBytes) { private byte[] joinRows(List<byte[]> rows, int totalBytes) {
if (StarRocksWriterOptions.StreamLoadFormat.CSV.equals(writerOptions.getStreamLoadFormat())) { if (StarRocksWriterOptions.StreamLoadFormat.CSV.equals(writerOptions.getStreamLoadFormat())) {
Map<String, Object> props = writerOptions.getLoadProps(); Map<String, Object> props = writerOptions.getLoadProps();
byte[] lineDelimiter = StarRocksDelimiterParser.parse((String)props.get("row_delimiter"), "\n").getBytes(StandardCharsets.UTF_8); byte[] lineDelimiter = StarRocksDelimiterParser.parse((String)props.get("row_delimiter"), "\n").getBytes(StandardCharsets.UTF_8);
ByteBuffer bos = ByteBuffer.allocate(totalBytes + rows.size() * lineDelimiter.length); ByteBuffer bos = ByteBuffer.allocate(totalBytes + rows.size() * lineDelimiter.length);
for (String row : rows) { for (byte[] row : rows) {
bos.put(row.getBytes(StandardCharsets.UTF_8)); bos.put(row);
bos.put(lineDelimiter); bos.put(lineDelimiter);
} }
return bos.array(); return bos.array();
@ -110,11 +110,11 @@ public class StarRocksStreamLoadVisitor {
bos.put("[".getBytes(StandardCharsets.UTF_8)); bos.put("[".getBytes(StandardCharsets.UTF_8));
byte[] jsonDelimiter = ",".getBytes(StandardCharsets.UTF_8); byte[] jsonDelimiter = ",".getBytes(StandardCharsets.UTF_8);
boolean isFirstElement = true; boolean isFirstElement = true;
for (String row : rows) { for (byte[] row : rows) {
if (!isFirstElement) { if (!isFirstElement) {
bos.put(jsonDelimiter); bos.put(jsonDelimiter);
} }
bos.put(row.getBytes(StandardCharsets.UTF_8)); bos.put(row);
isFirstElement = false; isFirstElement = false;
} }
bos.put("]".getBytes(StandardCharsets.UTF_8)); bos.put("]".getBytes(StandardCharsets.UTF_8));

7
starrockswriter/src/main/java/com/starrocks/connector/datax/plugin/writer/starrockswriter/manager/StarRocksWriterManager.java
View File

@ -20,7 +20,7 @@ public class StarRocksWriterManager {
private final StarRocksStreamLoadVisitor starrocksStreamLoadVisitor; private final StarRocksStreamLoadVisitor starrocksStreamLoadVisitor;
private final StarRocksWriterOptions writerOptions; private final StarRocksWriterOptions writerOptions;
private final List<String> buffer = new ArrayList<>(); private final List<byte[]> buffer = new ArrayList<>();
private int batchCount = 0; private int batchCount = 0;
private long batchSize = 0; private long batchSize = 0;
private volatile boolean closed = false; private volatile boolean closed = false;
@ -37,9 +37,10 @@ public class StarRocksWriterManager {
public final synchronized void writeRecord(String record) throws IOException { public final synchronized void writeRecord(String record) throws IOException {
checkFlushException(); checkFlushException();
try { try {
buffer.add(record); byte[] bts = record.getBytes(StandardCharsets.UTF_8);
buffer.add(bts);
batchCount++; batchCount++;
batchSize += record.getBytes(StandardCharsets.UTF_8).length; batchSize += bts.length;
if (batchCount >= writerOptions.getBatchRows() || batchSize >= writerOptions.getBatchSize()) { if (batchCount >= writerOptions.getBatchRows() || batchSize >= writerOptions.getBatchSize()) {
String label = createBatchLabel(); String label = createBatchLabel();
LOG.debug(String.format("StarRocks buffer Sinking triggered: rows[%d] label[%s].", batchCount, label)); LOG.debug(String.format("StarRocks buffer Sinking triggered: rows[%d] label[%s].", batchCount, label));