列更新

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/service/impl/KuaiShouAdSocialServiceImpl.java

@@ -9,18 +9,24 @@ import cn.com.ctop.job.kuaishou.data.utils.AlertUtils;
 import cn.com.ctop.job.kuaishou.data.utils.Check;
 import cn.com.ctop.job.kuaishou.data.utils.DateUtils;
 import cn.com.ctop.job.kuaishou.data.utils.HttpUtils;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 @Slf4j
 @Service
 public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+    @Autowired
     private KuaiShouAdSocialMapper kuaiShouAdSocialMapper;
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
@@ -72,11 +78,8 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         }
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertTaskList(addList);
-            try {
-                kuaiShouAdSocialMapper.insertTaskListSD(addList);
-            } catch (Exception e) {
-                log.error("insertTaskListSD", e);
-            }
+            //kuaiShouAdSocialMapper.insertTaskListSD(addList);
+            dorisLoadManager.add("media_api", "kuaishou_ad_social_task", addList);
         }
         if (page < totalPage) {
             getTaskList(accessToken, advertiserId, startDate, endDate, page + 1, isLog, errId, isSend);
@@ -143,11 +146,8 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         }
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertOrderList(addList);
-            try {
-                kuaiShouAdSocialMapper.insertOrderListSD(addList);
-            } catch (Exception e) {
-                log.error("insertOrderListSD", e);
-            }
+            //kuaiShouAdSocialMapper.insertOrderListSD(addList);
+            dorisLoadManager.add("media_api", "kuaishou_ad_social_task_order", addList);
         }
         if (page < totalPage) {
             getOrderList(accessToken, advertiserId, taskId, startDate, endDate, page + 1, isLog, errId, isSend);
@@ -221,11 +221,8 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         }
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertOrderDailyData(addList);
-            try {
-                kuaiShouAdSocialMapper.insertOrderDailyDataSD(addList);
-            } catch (Exception e) {
-                log.error("insertOrderDailyDataSD", e);
-            }
+            //kuaiShouAdSocialMapper.insertOrderDailyDataSD(addList);
+            dorisLoadManager.add("media_api", "kuaishou_ad_social_task_order_report_daily", addList);
         }
 
         log.info("【{} {} {}】录入快手星任务（视频）-订单分天数据完成 {} {}", advertiserId, taskId, orderIds, startDate, endDate);
@@ -306,7 +303,8 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
             }
         }
         if (!Check.isNull(addList)) {
-            kuaiShouAdSocialMapper.insertContractorMissionList(addList);
+            //kuaiShouAdSocialMapper.insertContractorMissionList(addList);
+            dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission", addList, "mission_id,contractor_user_id,advertiser_id,status");
         }
         if (offset + pageSize < totalCount) {
             getContractorMissionList(accessToken, advertiserId, contractorId, offset + pageSize);
@@ -345,7 +343,10 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         KuaiShouAdSocialContractorMission mission = JSONObject.toJavaObject(data, KuaiShouAdSocialContractorMission.class);
         mission.setAdvertiserId(advertiserId);
         mission.setContractorUserId(contractorId);
-        kuaiShouAdSocialMapper.insertContractorMissionInfo(Collections.singletonList(mission));
+        //kuaiShouAdSocialMapper.insertContractorMissionInfo(Collections.singletonList(mission));
+        dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission", mission,
+                "mission_id,contractor_user_id,advertiser_id, mission_name, mission_requirement," +
+                        "logo_url, uint_price_type, contractor_commission_ratio, uint_price_amount, progress_rate, budget");
         log.info("【{}】获取快手团长任务详情完成 {} {}", advertiserId, contractorId, missionId);
     }
 
@@ -391,7 +392,8 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
             }
         }
         if (!Check.isNull(addList)) {
-            kuaiShouAdSocialMapper.insertContractorMissionOrder(addList);
+            //kuaiShouAdSocialMapper.insertContractorMissionOrder(addList);
+            dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission_order", addList);
         }
         if (offset + pageSize < totalCount) {
             getContractorMissionOrder(accessToken, advertiserId, contractorId, missionId, offset + pageSize);

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/utils/JsonUtils.java

@@ -1,6 +1,7 @@
 package cn.com.ctop.job.kuaishou.data.utils;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.PropertyNamingStrategy;
 import com.alibaba.fastjson.serializer.NameFilter;
 import com.alibaba.fastjson.serializer.SerializeConfig;
@@ -50,7 +51,11 @@ public class JsonUtils {
     /**
      * 将对象转为 JSON 字符串，字段名转为下划线命名
      */
-    public static String toFilterJSON(Object obj) {
+    public static String toFilterJSONString(Object obj) {
         return JSON.toJSONString(obj, CAMEL_TO_SNAKE_FILTER, SerializerFeature.WriteMapNullValue, SerializerFeature.MapSortField);
     }
+
+    public static JSONObject toFilterJSON(Object obj) {
+        return JSON.parseObject(toFilterJSONString(obj));
+    }
 }

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/utils/doris/sink/DorisMultiTableStreamLoadManager.java

@@ -41,6 +41,20 @@ public class DorisMultiTableStreamLoadManager {
         }
     }
 
+    public void add(String database, String table, Object obj, String updateColumns) {
+        TableKey key = new TableKey(database, table);
+        if (obj instanceof Collection) {
+            Collection<?> collection = (Collection<?>) obj;
+            for (Object o : collection) {
+                bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
+                        .add(new RecordWithMeta(database, table, o, updateColumns));
+            }
+        } else {
+            bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
+                    .add(new RecordWithMeta(database, table, obj, updateColumns));
+        }
+    }
+
 
     @PreDestroy
     public void shutdown() {

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/utils/doris/sink/RecordWithMeta.java

@@ -14,4 +14,11 @@ public class RecordWithMeta {
     private String database;
     private String table;
     private Object data;
+    private String updateColumns;
+
+    public RecordWithMeta(String database, String table, Object data) {
+        this.database = database;
+        this.table = table;
+        this.data = data;
+    }
 }

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

@@ -3,6 +3,7 @@ package cn.com.ctop.job.kuaishou.utils.doris.sink;
 import cn.com.ctop.job.kuaishou.data.utils.JsonUtils;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpEntity;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPut;
@@ -40,18 +41,18 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
         RecordWithMeta sample = records.get(0);
         String db = sample.getDatabase();
         String table = sample.getTable();
-
+        String updateColumns = sample.getUpdateColumns();
         String label = labelPrefix + "_" + db + "_" + table + "_" + UUID.randomUUID();
         String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
 
         // 提取 data
         List<String> dataList = records.stream()
-                .map(d->{
+                .map(d -> {
                     Object data = d.getData();
                     if (data instanceof JSONObject) {
                         return data.toString();
-                    }else {
-                        return JsonUtils.toFilterJSON(data);
+                    } else {
+                        return JsonUtils.toFilterJSONString(data);
                     }
                 })
                 .collect(Collectors.toList());
@@ -67,6 +68,10 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
             String auth = Base64.getEncoder().encodeToString((user + ":" + password).getBytes(StandardCharsets.UTF_8));
             put.setHeader("Authorization", "Basic " + auth);
         }
+        if (StringUtils.isNotBlank(updateColumns)) {
+            put.setHeader("partial_columns", "true");
+            put.setHeader("columns", updateColumns);
+        }
 
         put.setHeader("format", format); // json
         put.setHeader("strip_outer_array", "true"); // json array