Merge #40 into master from pingkang

BytedanceCdpProject 加字段

* pingkang: (2 commits)
  selectdb batch stream load
  BytedanceCdpProject 加字段

Signed-off-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>
Merged-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>

CR-link: https://codeup.aliyun.com/60c06368dbbed34fe0f0c263/job/hr_job-bytedance/change/40

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/entity/BytedanceCdpProject.java

@@ -149,6 +149,9 @@ public class BytedanceCdpProject {
      */
     private Long microAppInstanceId;
 
+    private String deliveryMode;
+    private String bidType;
+
 
     /**
      * 创建时间
@@ -201,5 +204,7 @@ public class BytedanceCdpProject {
         if (!Check.isNull(optimizeGoal)) {
             this.externalAction = optimizeGoal.getString("external_action");
         }
+        this.deliveryMode = o.getString("delivery_mode");
+        this.bidType = o.getString("bid_type");
     }
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris/sink/SimpleHttpDorisSinkClient.java

@@ -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);