Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Improve] DataFrame CSV Stream Load optimization #137

Merged
merged 7 commits into from
Nov 7, 2023
Merged

[Improve] DataFrame CSV Stream Load optimization #137

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
7421fe4
当没有传columns时，默认以df schema
huanccwang Sep 7, 2023
dc53a50
增加判空
huanccwang Sep 7, 2023
a61cc0b
Merge remote-tracking branch 'origin/resolve'
huanccwang Oct 18, 2023
c6547fb
resolve conflict
huanccwang Oct 18, 2023
879a1d8
Merge branch 'resolve'
huanccwang Oct 25, 2023
3ce2d99
修改判空方式
huanccwang Oct 25, 2023
f9d14a9
Merge branch 'resolve'
huanccwang Oct 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;
import java.util.stream.Collectors;


/**
Expand Down Expand Up @@ -162,12 +163,17 @@ protected boolean isRedirectable(String method) {
return httpClientBuilder.build();
}

private HttpPut getHttpPut(String label, String loadUrlStr, Boolean enable2PC) {
private HttpPut getHttpPut(String label, String loadUrlStr, Boolean enable2PC, StructType schema) {
HttpPut httpPut = new HttpPut(loadUrlStr);
addCommonHeader(httpPut);
httpPut.setHeader("label", label);
if (StringUtils.isNotBlank(columns)) {
httpPut.setHeader("columns", columns);
} else {
if (schema != null && !schema.isEmpty()) {
String dfColumns = Arrays.stream(schema.fieldNames()).collect(Collectors.joining(","));
httpPut.setHeader("columns", dfColumns);
}
}
if (StringUtils.isNotBlank(maxFilterRatio)) {
httpPut.setHeader("max_filter_ratio", maxFilterRatio);
Expand Down Expand Up @@ -210,7 +216,7 @@ public long load(Iterator<InternalRow> rows, StructType schema)
try (CloseableHttpClient httpClient = getHttpClient()) {
String loadUrlStr = String.format(loadUrlPattern, getBackend(), db, tbl);
this.loadUrlStr = loadUrlStr;
HttpPut httpPut = getHttpPut(label, loadUrlStr, enable2PC);
HttpPut httpPut = getHttpPut(label, loadUrlStr, enable2PC, schema);
RecordBatchInputStream recodeBatchInputStream = new RecordBatchInputStream(RecordBatch.newBuilder(rows)
.format(fileType)
.sep(FIELD_DELIMITER)
Expand Down