Skip to content
Permalink
Browse files
Convert from arrow to rowbatch (#2723) 
For #2722
In our test environment, Doris cluster used 1 fe and 7 be (32C+128G). When using spakr-doris connecter to query a table containing 67 columns, it took about 1 hour for the query to return 69 million rows of data. After the improvement, the same query condition took 2.5 minutes and the query performance was significantly improved
  • Loading branch information
@Youngwb @imay
Youngwb authored and imay committed Jan 10, 2020
1 parent 817c819 commit 921a2caa836f1621557e3d56d67689e05f64786e
Showing 1 changed file with 96 additions and 40 deletions.
136 src/main/java/org/apache/doris/spark/serialization/RowBatch.java
View file
@@ -73,6 +73,7 @@ public void put(Object o) {
private int offsetInOneBatch = 0;
private int rowCountInOneBatch = 0;
private int readRowCount = 0;
private List<Row> rowBatch = new ArrayList<>();
private final ArrowStreamReader arrowStreamReader;
private final VectorSchemaRoot root;
private List<FieldVector> fieldVectors;
@@ -115,6 +116,11 @@ public boolean hasNext() throws DorisException {
}
offsetInOneBatch = 0;
rowCountInOneBatch = root.getRowCount();
// init the rowBatch
for (int i = 0; i < rowCountInOneBatch; ++i) {
rowBatch.add(new Row(fieldVectors.size()));
}
convertArrowToRowBatch();
return true;
}
} catch (IOException e) {
@@ -128,98 +134,135 @@ public boolean hasNext() throws DorisException {
return false;
}

public List<Object> next() throws DorisException {
private void addValueToRow(int rowIndex, Object obj) {
if (rowIndex > rowCountInOneBatch) {
String errMsg = "Get row offset: " + rowIndex + " larger than row size: " +
rowCountInOneBatch;
logger.error(errMsg);
throw new NoSuchElementException(errMsg);
}
rowBatch.get(rowIndex).put(obj);
}

public void convertArrowToRowBatch() throws DorisException {
try {
if (!hasNext()) {
String errMsg = "Get row offset:" + offsetInOneBatch + " larger than row size: " + rowCountInOneBatch;
logger.error(errMsg);
throw new NoSuchElementException(errMsg);
}
Row row = new Row(fieldVectors.size());
for (int j = 0; j < fieldVectors.size(); j++) {
FieldVector curFieldVector = fieldVectors.get(j);
for (int col = 0; col < fieldVectors.size(); col++) {
FieldVector curFieldVector = fieldVectors.get(col);
Types.MinorType mt = curFieldVector.getMinorType();
if (curFieldVector.isNull(offsetInOneBatch)) {
row.put(null);
continue;
}

final String currentType = schema.get(j).getType();
final String currentType = schema.get(col).getType();
switch (currentType) {
case "NULL_TYPE":
row.put(null);
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
addValueToRow(rowIndex, null);
}
break;
case "BOOLEAN":
Preconditions.checkArgument(mt.equals(Types.MinorType.BIT),
typeMismatchMessage(currentType, mt));
BitVector bitVector = (BitVector) curFieldVector;
int bit = bitVector.get(offsetInOneBatch);
row.put(bit != 0);
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = bitVector.isNull(rowIndex) ? null : bitVector.get(rowIndex) != 0;
addValueToRow(rowIndex, fieldValue);
}
break;
case "TINYINT":
Preconditions.checkArgument(mt.equals(Types.MinorType.TINYINT),
typeMismatchMessage(currentType, mt));
TinyIntVector tinyIntVector = (TinyIntVector) curFieldVector;
row.put(tinyIntVector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = tinyIntVector.isNull(rowIndex) ? null : tinyIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "SMALLINT":
Preconditions.checkArgument(mt.equals(Types.MinorType.SMALLINT),
typeMismatchMessage(currentType, mt));
SmallIntVector smallIntVector = (SmallIntVector) curFieldVector;
row.put(smallIntVector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = smallIntVector.isNull(rowIndex) ? null : smallIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "INT":
Preconditions.checkArgument(mt.equals(Types.MinorType.INT),
typeMismatchMessage(currentType, mt));
IntVector intVector = (IntVector) curFieldVector;
row.put(intVector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = intVector.isNull(rowIndex) ? null : intVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "BIGINT":
Preconditions.checkArgument(mt.equals(Types.MinorType.BIGINT),
typeMismatchMessage(currentType, mt));
BigIntVector bigIntVector = (BigIntVector) curFieldVector;
row.put(bigIntVector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = bigIntVector.isNull(rowIndex) ? null : bigIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "FLOAT":
Preconditions.checkArgument(mt.equals(Types.MinorType.FLOAT4),
typeMismatchMessage(currentType, mt));
Float4Vector float4Vector = (Float4Vector) curFieldVector;
row.put(float4Vector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = float4Vector.isNull(rowIndex) ? null : float4Vector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "TIME":
case "DOUBLE":
Preconditions.checkArgument(mt.equals(Types.MinorType.FLOAT8),
typeMismatchMessage(currentType, mt));
Float8Vector float8Vector = (Float8Vector) curFieldVector;
row.put(float8Vector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = float8Vector.isNull(rowIndex) ? null : float8Vector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "BINARY":
Preconditions.checkArgument(mt.equals(Types.MinorType.VARBINARY),
typeMismatchMessage(currentType, mt));
VarBinaryVector varBinaryVector = (VarBinaryVector) curFieldVector;
row.put(varBinaryVector.get(offsetInOneBatch));
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = varBinaryVector.isNull(rowIndex) ? null : varBinaryVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "DECIMAL":
Preconditions.checkArgument(mt.equals(Types.MinorType.VARCHAR),
typeMismatchMessage(currentType, mt));
VarCharVector varCharVectorForDecimal = (VarCharVector) curFieldVector;
String decimalValue = new String(varCharVectorForDecimal.get(offsetInOneBatch));
Decimal decimal = new Decimal();
try {
decimal.set(new scala.math.BigDecimal(new BigDecimal(decimalValue)));
} catch (NumberFormatException e) {
String errMsg = "Decimal response result '" + decimalValue + "' is illegal.";
logger.error(errMsg, e);
throw new DorisException(errMsg);
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (varCharVectorForDecimal.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String decimalValue = new String(varCharVectorForDecimal.get(rowIndex));
Decimal decimal = new Decimal();
try {
decimal.set(new scala.math.BigDecimal(new BigDecimal(decimalValue)));
} catch (NumberFormatException e) {
String errMsg = "Decimal response result '" + decimalValue + "' is illegal.";
logger.error(errMsg, e);
throw new DorisException(errMsg);
}
addValueToRow(rowIndex, decimal);
}
row.put(decimal);
break;
case "DECIMALV2":
Preconditions.checkArgument(mt.equals(Types.MinorType.DECIMAL),
typeMismatchMessage(currentType, mt));
DecimalVector decimalVector = (DecimalVector) curFieldVector;
Decimal decimalV2 = Decimal.apply(decimalVector.getObject(offsetInOneBatch));
row.put(decimalV2);
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (decimalVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
Decimal decimalV2 = Decimal.apply(decimalVector.getObject(rowIndex));
addValueToRow(rowIndex, decimalV2);
}
break;
case "DATE":
case "DATETIME":
@@ -229,23 +272,36 @@ public List<Object> next() throws DorisException {
Preconditions.checkArgument(mt.equals(Types.MinorType.VARCHAR),
typeMismatchMessage(currentType, mt));
VarCharVector varCharVector = (VarCharVector) curFieldVector;
String value = new String(varCharVector.get(offsetInOneBatch));
row.put(value);
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (varCharVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String value = new String(varCharVector.get(rowIndex));
addValueToRow(rowIndex, value);
}
break;
default:
String errMsg = "Unsupported type " + schema.get(j).getType();
String errMsg = "Unsupported type " + schema.get(col).getType();
logger.error(errMsg);
throw new DorisException(errMsg);
}
}
offsetInOneBatch++;
return row.getCols();
} catch (Exception e) {
close();
throw e;
}
}

public List<Object> next() throws DorisException {
if (!hasNext()) {
String errMsg = "Get row offset:" + offsetInOneBatch + " larger than row size: " + rowCountInOneBatch;
logger.error(errMsg);
throw new NoSuchElementException(errMsg);
}
return rowBatch.get(offsetInOneBatch++).getCols();
}

private String typeMismatchMessage(final String sparkType, final Types.MinorType arrowType) {
final String messageTemplate = "Spark type is %1$s, but arrow type is %2$s.";
return String.format(messageTemplate, sparkType, arrowType.name());

0 comments on commit 921a2ca

Please sign in to comment.