selectdb batch stream load

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/ByteDanceFlowDetailsReportServiceImpl.java

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.service.IByteDanceFlowDetailsReportService
 import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -32,6 +33,8 @@ import java.util.Map;
 @Service
 @Slf4j
 public class ByteDanceFlowDetailsReportServiceImpl implements IByteDanceFlowDetailsReportService {
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
     @Resource
     private ByteDanceFlowDetailsReportMapper byteDanceFlowDetailsReportMapper;
@@ -114,16 +117,13 @@ public class ByteDanceFlowDetailsReportServiceImpl implements IByteDanceFlowDeta
                                 .transferOut(transfer_out)
                                 .build();
                         dailies.add(daily);
+                        dorisLoadManager.add("media_api","bytedance_flow_details_report_daily", daily);
                     }
 
                     if (!Check.isNull(dailies)) {
                         log.info("账户id:{}，流水日报请求结果返回成功", advertiserId);
                         byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailies(dailies);
-                        try {
-                            byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailiesSD(dailies);
-                        }catch (Exception e) {
-                            log.error("",e);
-                        }
+                        //byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailiesSD(dailies);
                     }
                 }
 

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceAccountReportServiceImpl.java

@@ -7,18 +7,14 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
-import cn.com.ctop.job.bytedance.utils.doris_sink.DorisMultiTableStreamLoadManager;
-import cn.com.ctop.job.bytedance.utils.doris_sink.RecordWithMeta;
+import cn.com.ctop.job.bytedance.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.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.PreDestroy;
-import java.text.DecimalFormat;
 import java.util.*;
 
 @Service

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceCdpProjectServiceImpl.java

@@ -5,7 +5,7 @@ import cn.com.ctop.job.bytedance.data.mapper.BytedanceCdpProjectMapper;
 import cn.com.ctop.job.bytedance.data.mapper.BytedanceRefreshTokenMapper;
 import cn.com.ctop.job.bytedance.data.service.IBytedanceCdpProjectService;
 import cn.com.ctop.job.bytedance.data.utils.*;
-import cn.com.ctop.job.bytedance.utils.doris_sink.DorisMultiTableStreamLoadManager;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceProductReportServiceImpl.java

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -24,6 +25,9 @@ import java.util.Map;
 @Slf4j
 public class BytedanceProductReportServiceImpl implements IBytedanceProductReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
 
     @Autowired
@@ -83,6 +87,7 @@ public class BytedanceProductReportServiceImpl implements IBytedanceProductRepor
                             BytedanceProductReportDaily reportDaily = JSONObject.parseObject(metricsJson.toJSONString(), BytedanceProductReportDaily.class);
                             reportDaily.setAdvertiserId(advertiserId);
                             reportDailies.add(reportDaily);
+                            dorisLoadManager.add("media_api","bytedance_product_report_daily", reportDaily);
                         }
                     }
                 }
@@ -93,11 +98,7 @@ public class BytedanceProductReportServiceImpl implements IBytedanceProductRepor
             }
             if (!Check.isNull(reportDailies)) {
                 productReportMapper.replaceProductReportDailies(reportDailies);
-                try {
-                    productReportMapper.replaceProductReportDailiesSD(reportDailies);
-                }catch (Exception e){
-                    log.error("",e);
-                }
+                //productReportMapper.replaceProductReportDailiesSD(reportDailies);
                 log.info("账户ID：{}，产品数据日报请求成功,开始日期：{}，结束日期：{}，page:{}", advertiserId, startDate, endDate, page);
             }
             if (page < totalPage) {

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedancePromoterReportServiceImpl.java

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
 import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -26,14 +27,10 @@ import java.util.*;
 @EnableScheduling
 public class BytedancePromoterReportServiceImpl implements IBytedancePromoterReportService {
     @Autowired
-    private BytedancePromoterReportMapper promoterReportMapper;
-    /*private static final int MAX_BATCH_SIZE = 10000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
-    private final List<BytedancePromoterReportHour> promoterReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<BytedancePromoterReportDaily> promoterReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long promoterReportHourLastFlushTime = System.currentTimeMillis();
-    private long promoterReportDailyLastFlushTime = System.currentTimeMillis();*/
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
+    @Autowired
+    private BytedancePromoterReportMapper promoterReportMapper;
 
     @Override
     public List<JSONObject> getAccountList() {
@@ -90,6 +87,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             bytedancePromoterReportHour.setStatTimeDay(statTimeDay);
                             bytedancePromoterReportHour.setHour(hour);
                             hours.add(bytedancePromoterReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_promoter_report_hour", bytedancePromoterReportHour);
                         }
                     }
                 }
@@ -100,8 +99,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                promoterReportMapper.replacePromoterReportHourSD(hours);
-                //promoterReportHourBuffer.addAll(hours);
+                //promoterReportMapper.replacePromoterReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -171,6 +169,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             promoterReportDaily.setCdpPromotionName(cdpPromotionName);
                             promoterReportDaily.setStatTimeDay(statTimeDay);
                             promoterReportDailies.add(promoterReportDaily);
+                            dorisLoadManager.add("media_api", "bytedance_promoter_report_daily", promoterReportDaily);
                         }
                     }
                 }
@@ -181,8 +180,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                //promoterReportDailyBuffer.addAll(promoterReportDailies);
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
@@ -265,6 +263,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             bytedancePromoterReportHour.setStatTimeDay(statTimeDay);
                             bytedancePromoterReportHour.setHour(hour);
                             hours.add(bytedancePromoterReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_promoter_report_hour", bytedancePromoterReportHour);
                         }
                     }
                 }
@@ -284,8 +284,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                promoterReportMapper.replacePromoterReportHourSD(hours);
-                //promoterReportHourBuffer.addAll(hours);
+                //promoterReportMapper.replacePromoterReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -359,6 +358,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             promoterReportDaily.setCdpPromotionName(cdpPromotionName);
                             promoterReportDaily.setStatTimeDay(statTimeDay);
                             promoterReportDailies.add(promoterReportDaily);
+
+                            dorisLoadManager.add("media_api", "bytedance_promoter_report_daily", promoterReportDaily);
                         }
                     }
                 }
@@ -378,8 +379,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                //promoterReportDailyBuffer.addAll(promoterReportDailies);
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
@@ -406,86 +406,4 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
         return promoterReportMapper.getAllList();
     }
 
-    /*@Scheduled(fixedRate = 11000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行广告报表定时批处理任务....");
-        synchronized (promoterReportHourBuffer) {
-            long now = System.currentTimeMillis();
-            // promoterReportHourBuffer.size() >= MAX_BATCH_SIZE || (now - promoterReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !promoterReportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed PromoterReportHour batch size: {}，执行时间：{}", promoterReportHourBuffer.size(),seconds);
-
-                    promoterReportHourBuffer.clear();
-                    promoterReportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (promoterReportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            // promoterReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - promoterReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !promoterReportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed PromoterReportDaily batch size: {}，执行时间：{}", promoterReportDailyBuffer.size(),seconds);
-
-                    promoterReportDailyBuffer.clear();
-                    promoterReportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启，开始执行广告报表批处理任务....");
-        synchronized (promoterReportHourBuffer) {
-            if (!promoterReportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed PromoterReportHour batch size: {}", promoterReportHourBuffer.size());
-                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
-                    promoterReportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (promoterReportDailyBuffer) {
-            if (!promoterReportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed promoterReportDaily batch size: {}", promoterReportDailyBuffer.size());
-                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
-                    promoterReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing promoterReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
-
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceXingTuServiceImpl.java

@@ -11,6 +11,7 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.DateUtils;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -29,6 +30,9 @@ import java.util.Map;
 @Slf4j
 public class BytedanceXingTuServiceImpl implements IBytedanceXingTuService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceXingTuMapper bytedanceXingTuMapper;
 
     @Autowired
@@ -387,19 +391,13 @@ public class BytedanceXingTuServiceImpl implements IBytedanceXingTuService {
 
             if (!Check.isNull(reports)) {
                 bytedanceXingTuMapper.insertReport(reports);
-                try {
-                    bytedanceXingTuMapper.insertReportSD(reports);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //bytedanceXingTuMapper.insertReportSD(reports);
+                dorisLoadManager.add("media_api", "bytedance_xingtu_task_report" ,reports);
             }
             if (!Check.isNull(dailyReports)) {
                 bytedanceXingTuMapper.insertReportDaily(dailyReports);
-                try {
-                    bytedanceXingTuMapper.insertReportDailySD(dailyReports);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //bytedanceXingTuMapper.insertReportDailySD(dailyReports);
+                dorisLoadManager.add("media_api", "bytedance_xingtu_task_report_daily" ,dailyReports);
             }
 
             if (hasMore) {

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisConfiguration.java

@@ -1,11 +1,15 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils;
 
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisBufferConfig;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisSinkClient;
+import cn.com.ctop.job.bytedance.utils.doris.sink.SimpleHttpDorisSinkClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration
-public class DorisConfiguration {
+public class DorisConfig {
     @Value("${spring.doris.fe-host}")
     private String feHost;
 

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisBufferConfig.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisMultiTableStreamLoadManager.java

@@ -1,10 +1,11 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 
 import javax.annotation.PreDestroy;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -29,17 +30,19 @@ public class DorisMultiTableStreamLoadManager {
 
     public void add(String database, String table, Object obj) {
         TableKey key = new TableKey(database, table);
-        bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
-                .add(new RecordWithMeta(database, table, obj));
-    }
-
-    public void add(String database, String table, Collection<Object> objs) {
-        for (Object obj : objs) {
-            add(database, table, obj);
+        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));
+            }
+        } else {
+            bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
+                    .add(new RecordWithMeta(database, table, obj));
         }
-
     }
 
+
     @PreDestroy
     public void shutdown() {
         log.info("Shutting down DorisMultiTableStreamLoadManager");

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisSinkClient.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import java.util.List;
 @FunctionalInterface

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/RecordWithMeta.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

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

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
 import lombok.extern.slf4j.Slf4j;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/StrictRedirectStrategy.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import org.apache.http.HttpEntityEnclosingRequest;
 import org.apache.http.HttpRequest;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/TableBufferManager.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/TableKey.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;