|
|
@@ -1,6 +1,7 @@
|
|
|
package cn.com.ctop.job.bytedance.utils.doris.sink;
|
|
|
|
|
|
import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.apache.http.HttpEntity;
|
|
|
import org.apache.http.client.methods.CloseableHttpResponse;
|
|
|
@@ -44,11 +45,18 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
|
|
|
String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
|
|
|
|
|
|
// 提取 data
|
|
|
- List<Object> dataList = records.stream()
|
|
|
- .map(RecordWithMeta::getData)
|
|
|
+ List<String> dataList = records.stream()
|
|
|
+ .map(d->{
|
|
|
+ Object data = d.getData();
|
|
|
+ if (data instanceof JSONObject) {
|
|
|
+ return data.toString();
|
|
|
+ }else {
|
|
|
+ return JsonUtils.toFilterJSON(data);
|
|
|
+ }
|
|
|
+ })
|
|
|
.collect(Collectors.toList());
|
|
|
|
|
|
- String jsonBody = JsonUtils.toFilterJSON(dataList);
|
|
|
+ String jsonBody = dataList.toString();
|
|
|
//log.info("jsonBody:{}", jsonBody);
|
|
|
|
|
|
HttpPut put = new HttpPut(url);
|