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

Merge pull request #1603 from hf200012/doriswriter-doc-fix

Browse Source 
doris writer write error
This commit is contained in:
Trafalgar 2022-11-15 18:03:30 +08:00 committed by GitHub
commit 3e13d6a788
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 29 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doriswriter/doc/doriswriter.md
View File

@ -166,7 +166,7 @@ DorisWriter 通过Doris原生支持Stream load方式导入数据， DorisWriter
```json
"loadProps": {
"column_separator": "\\x01",
"row_delimiter": "\\x02"
"line_delimiter": "\\x02"
}
```

7
doriswriter/src/main/java/com/alibaba/datax/plugin/writer/doriswriter/DorisStreamLoadObserver.java
View File

@ -3,6 +3,7 @@ package com.alibaba.datax.plugin.writer.doriswriter;
import com.alibaba.fastjson.JSON;
import org.apache.commons.codec.binary.Base64;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
@ -124,7 +125,7 @@ public class DorisStreamLoadObserver {
private byte[] addRows(List<byte[]> rows, int totalBytes) {
if (Keys.StreamLoadFormat.CSV.equals(options.getStreamLoadFormat())) {
Map<String, Object> props = (options.getLoadProps() == null ? new HashMap<> () : options.getLoadProps());
byte[] lineDelimiter = DelimiterParser.parse((String)props.get("row_delimiter"), "\n").getBytes(StandardCharsets.UTF_8);
byte[] lineDelimiter = DelimiterParser.parse((String)props.get("line_delimiter"), "\n").getBytes(StandardCharsets.UTF_8);
ByteBuffer bos = ByteBuffer.allocate(totalBytes + rows.size() * lineDelimiter.length);
for (byte[] row : rows) {
bos.put(row);
@ -161,6 +162,8 @@ public class DorisStreamLoadObserver {
});
try ( CloseableHttpClient httpclient = httpClientBuilder.build()) {
HttpPut httpPut = new HttpPut(loadUrl);
httpPut.removeHeaders(HttpHeaders.CONTENT_LENGTH);
httpPut.removeHeaders(HttpHeaders.TRANSFER_ENCODING);
List<String> cols = options.getColumns();
if (null != cols && !cols.isEmpty() && Keys.StreamLoadFormat.CSV.equals(options.getStreamLoadFormat())) {
httpPut.setHeader("columns", String.join(",", cols.stream().map(f -> String.format("`%s`", f)).collect(Collectors.toList())));
@ -172,7 +175,7 @@ public class DorisStreamLoadObserver {
}
httpPut.setHeader("Expect", "100-continue");
httpPut.setHeader("label", label);
httpPut.setHeader("Content-Type", "application/x-www-form-urlencoded");
httpPut.setHeader("two_phase_commit", "false");
httpPut.setHeader("Authorization", getBasicAuthHeader(options.getUsername(), options.getPassword()));
httpPut.setEntity(new ByteArrayEntity(data));
httpPut.setConfig(RequestConfig.custom().setRedirectsEnabled(true).build());