selectdb batch stream load

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

@@ -153,14 +153,14 @@ public class BytedanceCdpProject {
     /**
      * 创建时间
      */
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    private Date createTime;
+    //@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    //private Date createTime;
 
     /**
      * 更新时间
      */
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    private Date updateTime;
+    //@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    //private Date updateTime;
 
     public BytedanceCdpProject() {
     }

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

@@ -74,6 +74,9 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setStatTimeDay(statTimeDay);
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_hour", advertiserReportHour);
+
                         }
                     }
                 }
@@ -84,8 +87,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
-                accountReportMapper.replaceAccountReportHourSD(hours);
-                //advertiserReportHourBuffer.addAll(hours);
+                //accountReportMapper.replaceAccountReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -146,6 +148,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportDaily.setAdvertiserId(advertiserId);
                             advertiserReportDaily.setStatTimeDay(statTimeDay);
                             dailyList.add(advertiserReportDaily);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_daily", advertiserReportDaily);
                         }
                     }
                 }
@@ -156,8 +160,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                accountReportMapper.replaceAccountReportDailySD(dailyList);
-                //advertiserReportDailyBuffer.addAll(dailyList);
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -229,7 +232,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
 
-                            dorisLoadManager.add(new RecordWithMeta("media_api","bytedance_advertiser_report_hour",advertiserReportHour));
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_hour", advertiserReportHour);
                         }
                     } else {
                         log.info("账户id:{}，日期:{}广告时报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -317,6 +320,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportDaily.setAdvertiserId(advertiserId);
                             advertiserReportDaily.setStatTimeDay(statTimeDay);
                             dailyList.add(advertiserReportDaily);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_daily", advertiserReportDaily);
                         }
                     } else {
                         log.info("账户id:{}，日期:{}，广告日报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -339,8 +344,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                accountReportMapper.replaceAccountReportDailySD(dailyList);
-                //advertiserReportDailyBuffer.addAll(dailyList);
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -357,83 +361,4 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
         }
     }
 
-    /*@Scheduled(fixedRate = 12000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行账户报表定时批处理任务....");
-        synchronized (advertiserReportHourBuffer) {
-            long now = System.currentTimeMillis();
-            // advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||                    (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if (!advertiserReportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
-                    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 advertiserReportHour batch size: {}，执行时间：{}", advertiserReportHourBuffer.size(), seconds);
-                    advertiserReportHourBuffer.clear();
-                    advertiserReportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (advertiserReportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            // advertiserReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - advertiserReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if (!advertiserReportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
-                    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 advertiserReportDaily batch size: {}，执行时间：{}", advertiserReportDailyBuffer.size(), seconds);
-                    advertiserReportDailyBuffer.clear();
-                    advertiserReportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启，开始执行广告报表批处理任务....");
-        synchronized (advertiserReportHourBuffer) {
-            if (!advertiserReportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
-                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
-                    advertiserReportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (advertiserReportDailyBuffer) {
-            if (!advertiserReportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
-                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
-                    advertiserReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
 }

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

@@ -5,6 +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 com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -29,6 +30,9 @@ import java.util.*;
 @Service
 public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceCdpProjectMapper bytedanceCdpProjectMapper;
 
     @Autowired
@@ -109,15 +113,13 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray array = data.getJSONArray("list");
                     for (int i = 0; i < array.size(); i++) {
-                        checkPromotionInfo(array.getJSONObject(i), advertiserId);
+                        JSONObject jsonObject = array.getJSONObject(i);
+                        checkPromotionInfo(jsonObject, advertiserId);
+                        dorisLoadManager.add("media_api", "bytedance_promotion_list", jsonObject);
                     }
 
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedancePromotionListSD(array);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedancePromotionListSD(array);
                         if (bytedanceCdpProjectMapper.replaceBytedancePromotionList(array) > 0)
                             log.info("---账户（{}）广告更新成功", advertiserId);
                     }
@@ -201,13 +203,11 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         BytedanceCdpProject advertiserToolsSite = new BytedanceCdpProject(array.getJSONObject(i));
                         projectList.add(advertiserToolsSite);
+
+                        dorisLoadManager.add("media_api", "bytedance_cdp_project_list", advertiserToolsSite);
                     }
                     if (!Check.isNull(projectList)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceCdpProjectListSD(projectList);
-                        } catch (Exception e) {
-                            log.error("replaceBytedanceCdpProjectListSD", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedanceCdpProjectListSD(projectList);
                         if (bytedanceCdpProjectMapper.replaceBytedanceCdpProjectList(projectList) > 0)
                             log.info("---账户（{}）项目信息更新成功", advertiserId);
                     }
@@ -326,11 +326,8 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray list = data.getJSONArray("list");
                     if (!Check.isNull(list)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfoSD(list);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        dorisLoadManager.add("media_api", "bytedance_tools_asset_link_list", list);
+                        //bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfoSD(list);
                         if (bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfo(list) > 0)
                             log.info("---账户（{}）抖小详情信息更新成功", advertiserId);
                     }
@@ -394,13 +391,15 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray list = data.getJSONArray("list");
                     if (!Check.isNull(list)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceWxAppInfoSD(list, advertiserId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedanceWxAppInfoSD(list, advertiserId);
                         if (bytedanceCdpProjectMapper.replaceBytedanceWxAppInfo(list, advertiserId) > 0)
                             log.info("---账户（{}）微小信息更新成功", advertiserId);
+
+                        for (Object o : list) {
+                            JSONObject json = (JSONObject) o;
+                            json.put("advertiser_id", advertiserId);
+                            dorisLoadManager.add("media_api", "bytedance_wx_app_info", json);
+                        }
                     }
 
                 } else {
@@ -435,7 +434,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
     @Override
-    public void getAdqCreatives(Long advertiserId, String startDate,String endDate) {
+    public void getAdqCreatives(Long advertiserId, String startDate, String endDate) {
         try {
             String accessToken = tokenMapper.onlyGetADQToken(advertiserId);
             if (Check.isNull(accessToken)) {
@@ -507,13 +506,12 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         JSONObject object = array.getJSONObject(i);
                         getAppId(object);
+
+                        object.put("advertiser_id", accountId);
+                        dorisLoadManager.add("media_api", "adq_dynamic_createive_list", object);
                     }
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceAdqDynamicCreateivesSD(array, accountId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceAdqDynamicCreateivesSD(array, accountId);
                         if (bytedanceCdpProjectMapper.replaceAdqDynamicCreateives(array, accountId) > 0)
                             log.info("---账户（{}）广点通创意更新成功", accountId);
                     }
@@ -529,19 +527,19 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             }
         }
         if (nextPage) {
-            rebackadqCreativesVo(accountId, accessToken, startDate ,endDate, page + 1, isDeleted, 1);
+            rebackadqCreativesVo(accountId, accessToken, startDate, endDate, page + 1, isDeleted, 1);
         }
     }
 
     @Override
-    public void getAdqWechatPages(Long advertiserId,Long pageId) {
+    public void getAdqWechatPages(Long advertiserId, Long pageId) {
         try {
             String accessToken = tokenMapper.onlyGetADQToken(advertiserId);
             if (Check.isNull(accessToken)) {
                 log.error("账户（{}）accessToken is null", advertiserId);
                 return;
             }
-            this.getAdqWechatPagesVo(advertiserId,pageId, accessToken, 1, true, 1);
+            this.getAdqWechatPagesVo(advertiserId, pageId, accessToken, 1, true, 1);
         } catch (Exception e) {
             e.printStackTrace();
             log.error("账户（{}）获取原生落地页异常", e.getMessage());
@@ -549,7 +547,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
 
-    private void getAdqWechatPagesVo(Long accountId,Long pageId, String accessToken, Integer page, Boolean isDeleted, int count) throws Exception {
+    private void getAdqWechatPagesVo(Long accountId, Long pageId, String accessToken, Integer page, Boolean isDeleted, int count) throws Exception {
         boolean nextPage = false;
         Map<String, Object> param = new HashMap<>();
         long timestamp = Instant.now().getEpochSecond();
@@ -588,13 +586,12 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         JSONObject object = array.getJSONObject(i);
                         getPageAppId(object);
+
+                        object.put("advertiser_id", accountId);
+                        dorisLoadManager.add("media_api", "adq_dynamic_page_list", object);
                     }
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceAdqDynamicPagesSD(array, accountId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceAdqDynamicPagesSD(array, accountId);
                         if (bytedanceCdpProjectMapper.replaceAdqDynamicPages(array, accountId) > 0)
                             log.info("---账户（{}）广点通落地页更新成功", accountId);
                     }
@@ -603,14 +600,14 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             } else {
                 if (count <= 4) {
                     Thread.sleep(500);
-                    getAdqWechatPagesVo(accountId,pageId, accessToken, page, isDeleted, count + 1);
+                    getAdqWechatPagesVo(accountId, pageId, accessToken, page, isDeleted, count + 1);
                 } else {
                     log.error("账户（{}）获取动态创意失败，入参：{}\n，结果：{}", accountId, JSONObject.toJSONString(param), obj);
                 }
             }
         }
         if (nextPage) {
-            getAdqWechatPagesVo(accountId,pageId, accessToken, page + 1, isDeleted, 1);
+            getAdqWechatPagesVo(accountId, pageId, accessToken, page + 1, isDeleted, 1);
         }
     }
 

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

@@ -4,6 +4,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 
 import javax.annotation.PreDestroy;
+import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -17,8 +18,8 @@ public class DorisMultiTableStreamLoadManager {
     private final DorisBufferConfig defaultBufferConfig;
 
     public DorisMultiTableStreamLoadManager(int uploaderThreads,
-                                             DorisBufferConfig defaultBufferConfig,
-                                             DorisSinkClient sinkClient) {
+                                            DorisBufferConfig defaultBufferConfig,
+                                            DorisSinkClient sinkClient) {
         log.info("init DorisMultiTableStreamLoadManager");
         this.defaultBufferConfig = defaultBufferConfig;
         this.sinkClient = sinkClient;
@@ -26,10 +27,17 @@ public class DorisMultiTableStreamLoadManager {
                 new CustomizableThreadFactory("doris-stream-uploader-"));
     }
 
-    public void add(RecordWithMeta record) {
-        TableKey key = new TableKey(record.getDatabase(), record.getTable());
+    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(record);
+                .add(new RecordWithMeta(database, table, obj));
+    }
+
+    public void add(String database, String table, Collection<Object> objs) {
+        for (Object obj : objs) {
+            add(database, table, obj);
+        }
+
     }
 
     @PreDestroy