소스 검색

selectdb stream load

huangpingkang 10 달 전
부모
커밋
f8b969ea9f
17개의 변경된 파일546개의 추가작업 그리고 231개의 파일을 삭제
  1. 17 85
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAccountReportServiceImpl.java
  2. 9 11
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAdIncomeServiceImpl.java
  3. 12 11
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAssetReportServiceImpl.java
  4. 15 93
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqCreativeCreativeReportServiceImpl.java
  5. 6 5
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqFlowDetailsReportServiceImpl.java
  6. 27 26
      job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqRetryReportServiceImpl.java
  7. 56 0
      job-adq/src/main/java/cn/com/ctop/job/adq/data/utils/JsonUtils.java
  8. 49 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/DorisConfig.java
  9. 16 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisBufferConfig.java
  10. 51 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisMultiTableStreamLoadManager.java
  11. 7 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisSinkClient.java
  12. 17 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/RecordWithMeta.java
  13. 94 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/SimpleHttpDorisSinkClient.java
  14. 38 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/StrictRedirectStrategy.java
  15. 96 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/TableBufferManager.java
  16. 26 0
      job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/TableKey.java
  17. 10 0
      job-adq/src/main/resources/application.yml

+ 17 - 85
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAccountReportServiceImpl.java

@@ -7,7 +7,9 @@ import cn.com.ctop.job.adq.data.service.IAdqAccountReportService;
 import cn.com.ctop.job.adq.data.utils.Check;
 import cn.com.ctop.job.adq.data.utils.FieldsConfig;
 import cn.com.ctop.job.adq.data.utils.HttpUtils;
+import cn.com.ctop.job.adq.data.utils.JsonUtils;
 import cn.com.ctop.job.adq.utils.RandomStringGenerator;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -26,16 +28,13 @@ import java.util.*;
 @EnableScheduling
 public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private AdqErrorLogMapper adqErrorLogMapper;
     @Autowired
     private AdqAccountReportMapper accountReportMapper;
-    /*private static final int MAX_BATCH_SIZE = 5000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
 
-    private final List<AdqAdvertiserReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<AdqAdvertiserReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
-    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();*/
 
     @Override
     public void getAdqAccountHour(Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isLog) {
@@ -78,6 +77,11 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
                                 hourReport.setAdvertiserId(advertiserId);
                                 hourReport.setStatTimeDay(Long.valueOf(startDate.replace("-", "")));
                                 hourList.add(hourReport);
+
+                                String filterJSON = JsonUtils.toFilterJSON(hourReport);
+                                JSONObject json = JSONObject.parseObject(filterJSON);
+                                json.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api", "adq_advertiser_report_hour", json);
                             }
                         }
                     }
@@ -97,8 +101,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             }
             if (!Check.isNull(hourList)) {
                 accountReportMapper.addAccountHourList(hourList);
-                accountReportMapper.addAccountHourListSD(hourList);
-                //advertiserReportHourBuffer.addAll(hourList);
+                //accountReportMapper.addAccountHourListSD(hourList);
             }
             if (hourList.size() >= 1000) {
                 getAdqAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -150,6 +153,11 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
                                 dailyReport.setAdvertiserId(advertiserId);
                                 dailyReport.setStatTimeDay(Long.valueOf(dataJson.getString("date").replace("-", "")));
                                 dailyList.add(dailyReport);
+
+                                String filterJSON = JsonUtils.toFilterJSON(dailyReport);
+                                JSONObject json = JSONObject.parseObject(filterJSON);
+                                json.put("video_outer_play100_count",dailyReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api", "adq_advertiser_report_daily", json);
                             }
                         }
                     }
@@ -169,8 +177,7 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.addAccountDailyList(dailyList);
-                accountReportMapper.addAccountDailyListSD(dailyList);
-                //advertiserReportDailyBuffer.addAll(dailyList);
+                //accountReportMapper.addAccountDailyListSD(dailyList);
             }
             if (dailyList.size() >= 1000) {
                 getAdqAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -182,79 +189,4 @@ public class AdqAccountReportServiceImpl implements IAdqAccountReportService {
         }
     }
 
-
-    /*@Scheduled(fixedRate = 15000) // 每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();
-
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    accountReportMapper.addAccountHourListSD(new ArrayList<>(advertiserReportHourBuffer));
-                    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 {
-                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
-                    accountReportMapper.addAccountDailyListSD(new ArrayList<>(advertiserReportDailyBuffer));
-                    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.addAccountHourListSD(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.addAccountDailyListSD(new ArrayList<>(advertiserReportDailyBuffer));
-                    advertiserReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
 }

+ 9 - 11
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAdIncomeServiceImpl.java

@@ -8,6 +8,7 @@ import cn.com.ctop.job.adq.data.service.IAdqAdIncomeService;
 import cn.com.ctop.job.adq.data.utils.AlertUtils;
 import cn.com.ctop.job.adq.data.utils.Check;
 import cn.com.ctop.job.adq.data.utils.HttpUtils;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -25,6 +26,8 @@ import java.util.Map;
 @Service
 @Slf4j
 public class AdqAdIncomeServiceImpl implements IAdqAdIncomeService {
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
     @Resource
     private RedisTemplate redis2Template_db5;
     @Resource
@@ -86,6 +89,8 @@ public class AdqAdIncomeServiceImpl implements IAdqAdIncomeService {
                             report.setAdSlotName(getAdSlotName(report.getAdSlot()));
                             //System.out.println(report);
                             reports.add(report);
+
+                            dorisLoadManager.add("media_api","adq_publisher_adunit_general", report);
                         }
                     }
 
@@ -101,11 +106,7 @@ public class AdqAdIncomeServiceImpl implements IAdqAdIncomeService {
             }
             if (!Check.isNull(reports)) {
                 adqAdIncomeMapper.insertAdqAdIncome(reports);
-                try{
-                    adqAdIncomeMapper.insertAdqAdIncomeSD(reports);
-                }catch (Exception e){
-                    log.error("insertAdqAdIncomeSD",e);
-                }
+                //adqAdIncomeMapper.insertAdqAdIncomeSD(reports);
             }
 
             if (totalPage > page) {
@@ -169,6 +170,8 @@ public class AdqAdIncomeServiceImpl implements IAdqAdIncomeService {
                             report.setAdSlotName(getAdSlotName(report.getAdSlot()));
                             //System.out.println(report);
                             reports.add(report);
+
+                            dorisLoadManager.add("media_api","adq_publisher_adunit_general",report);
                         }
                     }
 
@@ -184,12 +187,7 @@ public class AdqAdIncomeServiceImpl implements IAdqAdIncomeService {
             }
             if (!Check.isNull(reports)) {
                 adqAdIncomeMapper.insertAdqAdIncome(reports);
-                try {
-                    adqAdIncomeMapper.insertAdqAdIncomeSD(reports);
-                }catch (Exception e) {
-                    log.error("insertAdqAdIncomeSD",e);
-                }
-
+                //adqAdIncomeMapper.insertAdqAdIncomeSD(reports);
             }
 
             if (totalPage > page) {

+ 12 - 11
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqAssetReportServiceImpl.java

@@ -9,7 +9,9 @@ import cn.com.ctop.job.adq.data.service.IAdqAssetReportService;
 import cn.com.ctop.job.adq.data.utils.AlertUtils;
 import cn.com.ctop.job.adq.data.utils.Check;
 import cn.com.ctop.job.adq.data.utils.HttpUtils;
+import cn.com.ctop.job.adq.data.utils.JsonUtils;
 import cn.com.ctop.job.adq.utils.RandomStringGenerator;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -23,6 +25,8 @@ import java.util.*;
 @Service
 public class AdqAssetReportServiceImpl implements IAdqAssetReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+    @Autowired
     private AdqErrorLogMapper adqErrorLogMapper;
     @Autowired
     private AdqAssetReportMapper adqAssetReportMapper;
@@ -66,6 +70,11 @@ public class AdqAssetReportServiceImpl implements IAdqAssetReportService {
                                 report.setStatTimeDay(Integer.valueOf(dataJson.getString("date").replace("-", "")));
                                 dailies.add(report);
 
+                                String filterJSON = JsonUtils.toFilterJSON(report);
+                                JSONObject filterJsonObject = JSONObject.parseObject(filterJSON);
+                                filterJsonObject.put("video_outer_play100_count", report.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api", "adq_asset_report_daily", filterJsonObject);
+
                                 Long marketingAssetId = report.getMarketingAssetId();
                                 if (!Check.isNull(marketingAssetId)) {
                                     String marketingTargetType = report.getMarketingTargetType();
@@ -88,11 +97,7 @@ public class AdqAssetReportServiceImpl implements IAdqAssetReportService {
             }
             if (!Check.isNull(dailies)) {
                 adqAssetReportMapper.insertAdqAssetReportDaily(dailies);
-                try {
-                    adqAssetReportMapper.insertAdqAssetReportDailySD(dailies);
-                }catch (Exception e) {
-                    log.error("insertAdqAssetReportDailySD",e);
-                }
+                //adqAssetReportMapper.insertAdqAssetReportDailySD(dailies);
             }
             if (!Check.isNullMap(assetIdsMap)) {
                 getAdqAssetInfo(assetIdsMap, advertiserId, accessToken, 1, null, null);
@@ -216,12 +221,8 @@ public class AdqAssetReportServiceImpl implements IAdqAssetReportService {
                 }
                 if (!Check.isNull(infos)) {
                     adqAssetReportMapper.insertAdqAssetInfo(infos);
-                    try {
-                        adqAssetReportMapper.insertAdqAssetInfoSD(infos);
-                    }catch (Exception e) {
-                        log.error("insertAdqAssetInfoSD",e);
-                    }
-
+                    //adqAssetReportMapper.insertAdqAssetInfoSD(infos);
+                    dorisLoadManager.add("media_api", "adq_asset_info", infos);
                 }
 
                 if (page == totalPage) {

+ 15 - 93
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqCreativeCreativeReportServiceImpl.java

@@ -9,7 +9,9 @@ import cn.com.ctop.job.adq.data.service.IAdqCreativeReportService;
 import cn.com.ctop.job.adq.data.utils.Check;
 import cn.com.ctop.job.adq.data.utils.FieldsConfig;
 import cn.com.ctop.job.adq.data.utils.HttpUtils;
+import cn.com.ctop.job.adq.data.utils.JsonUtils;
 import cn.com.ctop.job.adq.utils.RandomStringGenerator;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -33,14 +35,8 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     @Autowired
     private AdqErrorLogMapper adqErrorLogMapper;
 
-
-    /*private static final int MAX_BATCH_SIZE = 5000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
-
-    private final List<AdqCreativeReportHour> creativeReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<AdqCreativeReportDaily> creativeReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long creativeReportHourLastFlushTime = System.currentTimeMillis();
-    private long creativeReportDailyLastFlushTime = System.currentTimeMillis();*/
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
 
     @Override
@@ -84,6 +80,10 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
                                 hourReport.setAdvertiserId(advertiserId);
                                 hourReport.setStatTimeDay(Long.valueOf(startDate.replace("-", "")));
                                 hourList.add(hourReport);
+
+                                JSONObject filterJson = JSONObject.parseObject(JsonUtils.toFilterJSON(hourReport));
+                                filterJson.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api","adq_creative_report_hour", filterJson);
                             }
                         }
                     }
@@ -104,9 +104,8 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             log.info("hourList.size:{}", hourList.size());
             if (!Check.isNull(hourList)) {
                 adqReportMapper.addsHourList(hourList);
-                adqReportMapper.addsHourListSD(hourList);
+                //adqReportMapper.addsHourListSD(hourList);
                 log.info("新增数据完成account:id:{}", advertiserId);
-                //creativeReportHourBuffer.addAll(hourList);
             }
             if (hourList.size() >= 1000) {
                 getAdqCreativeHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -165,6 +164,11 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
                                 hourReport.setAdvertiserId(advertiserId);
                                 hourReport.setStatTimeDay(Long.valueOf(dataJson.getString("date").replace("-", "")));
                                 dailies.add(hourReport);
+
+                                JSONObject filterJson = JSONObject.parseObject(JsonUtils.toFilterJSON(hourReport));
+                                filterJson.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api","adq_creative_report_daily", filterJson);
+
                             }
                         }
                     }
@@ -184,8 +188,7 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
             }
             if (!Check.isNull(dailies)) {
                 adqReportMapper.addsDaily(dailies);
-                adqReportMapper.addsDailySD(dailies);
-                //creativeReportDailyBuffer.addAll(dailies);
+                //adqReportMapper.addsDailySD(dailies);
             }
             if (dailies.size() >= 1000) {
                 getAdqCreativeDaily(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
@@ -206,85 +209,4 @@ public class AdqCreativeCreativeReportServiceImpl implements IAdqCreativeReportS
     public List<JSONObject> getAccountListByDate(Integer startDate, Integer endDate) {
         return adqErrorLogMapper.getAccountListByDate(startDate, endDate);
     }
-
-
-    /*@Scheduled(fixedRate = 10000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行创意报表定时批处理任务....");
-        synchronized (creativeReportHourBuffer) {
-            long now = System.currentTimeMillis();
-            //  creativeReportHourBuffer.size() >= MAX_BATCH_SIZE ||  (now - creativeReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if (!creativeReportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
-                    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 creativeReportHour batch size: {},执行时间:{}", creativeReportHourBuffer.size(), seconds);
-                    creativeReportHourBuffer.clear();
-                    creativeReportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (creativeReportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            // creativeReportDailyBuffer.size() >= MAX_BATCH_SIZE || (now - creativeReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !creativeReportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
-                    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 creativeReportDaily batch size: {},执行时间:{}", creativeReportDailyBuffer.size(),seconds);
-
-                    creativeReportDailyBuffer.clear();
-                    creativeReportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启,开始执行创意报表批处理任务....");
-        synchronized (creativeReportHourBuffer) {
-            if (!creativeReportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportHour batch size: {}", creativeReportHourBuffer.size());
-                    adqReportMapper.addsHourListSD(new ArrayList<>(creativeReportHourBuffer));
-                    creativeReportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (creativeReportDailyBuffer) {
-            if (!creativeReportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed creativeReportDaily batch size: {}", creativeReportDailyBuffer.size());
-                    adqReportMapper.addsDailySD(new ArrayList<>(creativeReportDailyBuffer));
-                    creativeReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportDaily batch", e);
-                }
-            }
-        }
-    }*/
 }

+ 6 - 5
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqFlowDetailsReportServiceImpl.java

@@ -10,6 +10,7 @@ import cn.com.ctop.job.adq.data.utils.Check;
 import cn.com.ctop.job.adq.data.utils.DateUtils;
 import cn.com.ctop.job.adq.data.utils.HttpUtils;
 import cn.com.ctop.job.adq.utils.RandomStringGenerator;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +32,8 @@ import java.util.stream.Collectors;
 @Service
 @Slf4j
 public class AdqFlowDetailsReportServiceImpl implements IAdqFlowDetailsReportService {
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
     @Resource
     private AdqFlowDetailsReportDailyMapper adqFlowDetailsReportDailyMapper;
@@ -114,11 +117,9 @@ public class AdqFlowDetailsReportServiceImpl implements IAdqFlowDetailsReportSer
                             dailyList = buildFlowDetailsReportDaily(dailyList);
 
                             adqFlowDetailsReportDailyMapper.replaceFlowDetailsReportDailies(dailyList);
-                            try {
-                                adqFlowDetailsReportDailyMapper.replaceFlowDetailsReportDailiesSD(dailyList);
-                            }catch (Exception e) {
-                                log.error("replaceFlowDetailsReportDailiesSD",e);
-                            }
+
+                            dorisLoadManager.add("media_api", "adq_flow_details_report_daily", dailyList);
+                            //adqFlowDetailsReportDailyMapper.replaceFlowDetailsReportDailiesSD(dailyList);
                         }
                     }
                 }

+ 27 - 26
job-adq/src/main/java/cn/com/ctop/job/adq/data/service/impl/AdqRetryReportServiceImpl.java

@@ -5,11 +5,9 @@ import cn.com.ctop.job.adq.data.mapper.AdqAccountReportMapper;
 import cn.com.ctop.job.adq.data.mapper.AdqCreativeReportMapper;
 import cn.com.ctop.job.adq.data.mapper.AdqErrorLogMapper;
 import cn.com.ctop.job.adq.data.service.IAdqRetryReportService;
-import cn.com.ctop.job.adq.data.utils.AlertUtils;
-import cn.com.ctop.job.adq.data.utils.Check;
-import cn.com.ctop.job.adq.data.utils.FieldsConfig;
-import cn.com.ctop.job.adq.data.utils.HttpUtils;
+import cn.com.ctop.job.adq.data.utils.*;
 import cn.com.ctop.job.adq.utils.RandomStringGenerator;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -28,6 +26,8 @@ import java.util.concurrent.Executors;
 @Service
 public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+    @Autowired
     private AdqCreativeReportMapper adqReportMapper;
     @Autowired
     private AdqAccountReportMapper accountReportMapper;
@@ -74,6 +74,10 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
                             hourReport.setAdvertiserId(advertiserId);
                             hourReport.setStatTimeDay(Long.valueOf(startDate.replace("-", "")));
                             hourList.add(hourReport);
+
+                            JSONObject filterJson = JSONObject.parseObject(JsonUtils.toFilterJSON(hourReport));
+                            filterJson.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                            dorisLoadManager.add("media_api","adq_creative_report_hour",filterJson);
                         }
                     }
                 }
@@ -84,12 +88,7 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
         }
         if (!Check.isNull(hourList)) {
             adqReportMapper.addsHourList(hourList);
-            try {
-                adqReportMapper.addsHourListSD(hourList);
-            }catch (Exception e) {
-                log.error("addsHourListSD",e);
-            }
-
+            //adqReportMapper.addsHourListSD(hourList);
         }
         if (hourList.size() >= 1000) {
             getCreativeHourRetry(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -144,6 +143,11 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
                                 hourReport.setAdvertiserId(advertiserId);
                                 hourReport.setStatTimeDay(Long.valueOf(dataJson.getString("date").replace("-", "")));
                                 dailies.add(hourReport);
+
+                                JSONObject filterJson = JSONObject.parseObject(JsonUtils.toFilterJSON(hourReport));
+                                filterJson.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api","adq_creative_report_daily", filterJson);
+
                             }
                         }
                     }
@@ -154,12 +158,7 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
             }
             if (!Check.isNull(dailies)) {
                 adqReportMapper.addsDaily(dailies);
-                try {
-                    adqReportMapper.addsDailySD(dailies);
-                }catch (Exception e) {
-                    log.error("addsDailySD",e);
-                }
-
+                //adqReportMapper.addsDailySD(dailies);
             }
             if (dailies.size() >= 1000) {
                 getCreativeDailyRetry(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -215,6 +214,11 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
                                 hourReport.setAdvertiserId(advertiserId);
                                 hourReport.setStatTimeDay(Long.valueOf(startDate.replace("-", "")));
                                 hourList.add(hourReport);
+
+                                String filterJSON = JsonUtils.toFilterJSON(hourReport);
+                                JSONObject json = JSONObject.parseObject(filterJSON);
+                                json.put("video_outer_play100_count",hourReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api", "adq_advertiser_report_hour", json);
                             }
                         }
                     }
@@ -225,11 +229,7 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
             }
             if (!Check.isNull(hourList)) {
                 accountReportMapper.addAccountHourList(hourList);
-                try {
-                    accountReportMapper.addAccountHourListSD(hourList);
-                }catch (Exception e){
-                    log.error("addAccountHourList",e);
-                }
+                //accountReportMapper.addAccountHourListSD(hourList);
             }
             if (hourList.size() >= 1000) {
                 getAccountHourRetry(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -284,6 +284,11 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
                                 dailyReport.setAdvertiserId(advertiserId);
                                 dailyReport.setStatTimeDay(Long.valueOf(dataJson.getString("date").replace("-", "")));
                                 dailyList.add(dailyReport);
+
+                                String filterJSON = JsonUtils.toFilterJSON(dailyReport);
+                                JSONObject json = JSONObject.parseObject(filterJSON);
+                                json.put("video_outer_play100_count",dailyReport.getVideoOuterPlay100Count());
+                                dorisLoadManager.add("media_api", "adq_advertiser_report_daily", json);
                             }
                         }
                     }
@@ -294,11 +299,7 @@ public class AdqRetryReportServiceImpl implements IAdqRetryReportService {
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.addAccountDailyList(dailyList);
-                try {
-                    accountReportMapper.addAccountDailyListSD(dailyList);
-                }catch (Exception e){
-                    log.error("addAccountDailyListSD",e);
-                }
+                //accountReportMapper.addAccountDailyListSD(dailyList);
             }
             if (dailyList.size() >= 1000) {
                 getAccountDailyRetry(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);

+ 56 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/data/utils/JsonUtils.java

@@ -0,0 +1,56 @@
+package cn.com.ctop.job.adq.data.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.PropertyNamingStrategy;
+import com.alibaba.fastjson.serializer.NameFilter;
+import com.alibaba.fastjson.serializer.SerializeConfig;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+
+import java.util.regex.Pattern;
+
+public class JsonUtils {
+
+    // 正则预编译
+    private static final Pattern INVALID_PREFIX = Pattern.compile("^[^a-zA-Z0-9]+");
+    private static final Pattern INVALID_SUFFIX = Pattern.compile("[^a-zA-Z0-9]+$");
+    private static final Pattern LETTER_NUMBER = Pattern.compile("([a-z])([0-9])");
+    //private static final Pattern NUMBER_LETTER = Pattern.compile("([0-9])([a-zA-Z])");
+    private static final Pattern CAMEL_CASE = Pattern.compile("([a-z])([A-Z])");
+
+    // 全局 SerializeConfig
+    private static final SerializeConfig SNAKE_CASE_CONFIG = new SerializeConfig();
+
+    static {
+        SNAKE_CASE_CONFIG.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
+    }
+
+    // NameFilter
+    private static final NameFilter CAMEL_TO_SNAKE_FILTER = (object, name, value) -> {
+        // 去除非法首尾字符
+        name = INVALID_PREFIX.matcher(name).replaceAll("");
+        name = INVALID_SUFFIX.matcher(name).replaceAll("");
+
+        // a3 -> a_3,3a -> 3_a
+        name = LETTER_NUMBER.matcher(name).replaceAll("$1_$2");
+        //name = NUMBER_LETTER.matcher(name).replaceAll("$1_$2");
+
+        // camelCase -> camel_case
+        name = CAMEL_CASE.matcher(name).replaceAll("$1_$2");
+
+        return name.toLowerCase();
+    };
+
+    /**
+     * 将对象转为 JSON 字符串,字段名转为下划线命名。
+     */
+    public static String toSnakeCaseJsonIgnoreNull(Object obj) {
+        return JSON.toJSONString(obj, SNAKE_CASE_CONFIG);
+    }
+
+    /**
+     * 将对象转为 JSON 字符串,字段名转为下划线命名
+     */
+    public static String toFilterJSON(Object obj) {
+        return JSON.toJSONString(obj, CAMEL_TO_SNAKE_FILTER, SerializerFeature.WriteMapNullValue, SerializerFeature.MapSortField);
+    }
+}

+ 49 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/DorisConfig.java

@@ -0,0 +1,49 @@
+package cn.com.ctop.job.adq.utils;
+
+import cn.com.ctop.job.adq.utils.doris.sink.DorisBufferConfig;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisMultiTableStreamLoadManager;
+import cn.com.ctop.job.adq.utils.doris.sink.DorisSinkClient;
+import cn.com.ctop.job.adq.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 DorisConfig {
+    @Value("${spring.doris.fe-host}")
+    private String feHost;
+
+    @Value("${spring.doris.user}")
+    private String userName;
+
+    @Value("${spring.doris.password}")
+    private String password;
+
+    @Value("${spring.doris.label-prefix}")
+    private String labelPrefix;
+
+    @Value("${spring.doris.sink-threads}")
+    private int uploaderThreads;
+
+    @Value("${spring.doris.sink-batch-rows}")
+    private int flushMaxRows;
+
+    @Value("${spring.doris.sink-batch-mb}")
+    private long flushMaxMb;
+
+    @Value("${spring.doris.sink-batch-seconds}")
+    private long flushIntervalSeconds;
+
+    @Bean
+    public DorisMultiTableStreamLoadManager dorisStreamLoadManager() {
+        DorisBufferConfig defaultConfig = DorisBufferConfig.builder()
+                .bufferFlushMaxRows(flushMaxRows)
+                .bufferFlushMaxBytes(flushMaxMb * 1024 * 1024)
+                .bufferFlushIntervalMs(flushIntervalSeconds * 1000)
+                .build();
+
+        DorisSinkClient sinkClient = new SimpleHttpDorisSinkClient(feHost, userName, password, labelPrefix);
+
+        return new DorisMultiTableStreamLoadManager(uploaderThreads, defaultConfig, sinkClient);
+    }
+}

+ 16 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisBufferConfig.java

@@ -0,0 +1,16 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DorisBufferConfig {
+    private long bufferFlushMaxRows;
+    private long bufferFlushMaxBytes;
+    private long bufferFlushIntervalMs;
+}

+ 51 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisMultiTableStreamLoadManager.java

@@ -0,0 +1,51 @@
+package cn.com.ctop.job.adq.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.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Slf4j
+public class DorisMultiTableStreamLoadManager {
+    private final Map<TableKey, TableBufferManager> bufferManagers = new ConcurrentHashMap<>();
+    private final ExecutorService uploaderExecutor;
+    private final DorisSinkClient sinkClient;
+    private final DorisBufferConfig defaultBufferConfig;
+
+    public DorisMultiTableStreamLoadManager(int uploaderThreads,
+                                            DorisBufferConfig defaultBufferConfig,
+                                            DorisSinkClient sinkClient) {
+        log.info("init DorisMultiTableStreamLoadManager");
+        this.defaultBufferConfig = defaultBufferConfig;
+        this.sinkClient = sinkClient;
+        this.uploaderExecutor = Executors.newFixedThreadPool(uploaderThreads,
+                new CustomizableThreadFactory("doris-stream-uploader-"));
+    }
+
+    public void add(String database, String table, Object obj) {
+        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));
+            }
+        } 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");
+        bufferManagers.values().forEach(TableBufferManager::flushNow);
+        uploaderExecutor.shutdown();
+    }
+}

+ 7 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/DorisSinkClient.java

@@ -0,0 +1,7 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import java.util.List;
+@FunctionalInterface
+public interface DorisSinkClient {
+    void streamLoad(List<RecordWithMeta> batch) throws Exception;
+}

+ 17 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/RecordWithMeta.java

@@ -0,0 +1,17 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class RecordWithMeta {
+    private String database;
+    private String table;
+    private Object data;
+}

+ 94 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/SimpleHttpDorisSinkClient.java

@@ -0,0 +1,94 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import cn.com.ctop.job.adq.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;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SimpleHttpDorisSinkClient implements DorisSinkClient {
+    private final String feHost;      // http://fe_host:8030
+    private final String user;        // doris 用户名
+    private final String password;    // doris 密码
+    private final String format;      // json/csv, 推荐 json
+    private final String labelPrefix; // stream load label 前缀(可选)
+
+    public SimpleHttpDorisSinkClient(String feHost, String user, String password, String labelPrefix) {
+        this.feHost = feHost;
+        this.user = user;
+        this.password = password;
+        this.format = "json";
+        this.labelPrefix = labelPrefix;
+    }
+
+    @Override
+    public void streamLoad(List<RecordWithMeta> records) throws Exception {
+        if (records == null || records.isEmpty()) return;
+
+        RecordWithMeta sample = records.get(0);
+        String db = sample.getDatabase();
+        String table = sample.getTable();
+
+        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->{
+                    Object data = d.getData();
+                    if (data instanceof JSONObject) {
+                        return data.toString();
+                    }else {
+                        return JsonUtils.toFilterJSON(data);
+                    }
+                })
+                .collect(Collectors.toList());
+
+        String jsonBody = dataList.toString();
+        //log.info("jsonBody:{}", jsonBody);
+
+        HttpPut put = new HttpPut(url);
+        put.setHeader("Expect", "100-continue");
+        put.setHeader("label", label);
+        put.setHeader("Content-Type", "application/json");
+        if (user != null && password != null) {
+            String auth = Base64.getEncoder().encodeToString((user + ":" + password).getBytes(StandardCharsets.UTF_8));
+            put.setHeader("Authorization", "Basic " + auth);
+        }
+
+        put.setHeader("format", format); // json
+        put.setHeader("strip_outer_array", "true"); // json array
+        put.setHeader("fuzzy_parse", "true"); // json array
+
+        put.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
+
+        try (CloseableHttpClient httpClient = HttpClients.custom()
+                .setRedirectStrategy(new StrictRedirectStrategy())  // 支持 307/301 重定向
+                .build();
+             CloseableHttpResponse response = httpClient.execute(put)) {
+            int code = response.getStatusLine().getStatusCode();
+            HttpEntity entity = response.getEntity();
+            String responseBody = EntityUtils.toString(entity, "UTF-8");
+            if (code != 200 && code != 201) {
+                throw new RuntimeException("Doris stream load failed. Status: " + code + ", response: " + responseBody);
+            }
+            // 可选: 解析返回内容判断 success
+            if (!responseBody.contains("\"Status\": \"Success\"")) {
+                throw new RuntimeException("Stream load returned non-success: " + responseBody);
+            }
+            log.info("Doris stream load success: {}", responseBody);
+        }
+    }
+}

+ 38 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/StrictRedirectStrategy.java

@@ -0,0 +1,38 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import org.apache.http.HttpEntityEnclosingRequest;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolException;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.protocol.HttpContext;
+
+import java.net.URI;
+
+public class StrictRedirectStrategy extends DefaultRedirectStrategy {
+
+    @Override
+    protected boolean isRedirectable(String method) {
+        // 允许 GET, POST, PUT 等都可重定向
+        return true;
+    }
+
+    @Override
+    public HttpUriRequest getRedirect(HttpRequest request, HttpResponse response, HttpContext context)
+            throws ProtocolException {
+        URI uri = getLocationURI(request, response, context);
+
+        String method = request.getRequestLine().getMethod();
+        if (method.equalsIgnoreCase(HttpPut.METHOD_NAME)) {
+            HttpPut put = new HttpPut(uri);
+            put.setHeaders(request.getAllHeaders());
+            if (request instanceof HttpEntityEnclosingRequest) {
+                put.setEntity(((HttpEntityEnclosingRequest) request).getEntity());
+            }
+            return put;
+        }
+        return super.getRedirect(request, response, context);
+    }
+}

+ 96 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/TableBufferManager.java

@@ -0,0 +1,96 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Slf4j
+public class TableBufferManager {
+
+    private final TableKey tableKey;
+    private final DorisBufferConfig config;
+    private final DorisSinkClient sinkClient;
+    private final ExecutorService uploaderExecutor;
+
+    private final List<RecordWithMeta> buffer = new ArrayList<>();
+    private final Object lock = new Object();
+    private final AtomicLong currentBytes = new AtomicLong(0);
+
+    private final ScheduledExecutorService scheduler =
+            Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("flusher-" + UUID.randomUUID()));
+
+    public TableBufferManager(TableKey tableKey,
+                              DorisBufferConfig config,
+                              DorisSinkClient sinkClient,
+                              ExecutorService uploaderExecutor) {
+        this.tableKey = tableKey;
+        this.config = config;
+        this.sinkClient = sinkClient;
+        this.uploaderExecutor = uploaderExecutor;
+
+        scheduler.scheduleAtFixedRate(this::tryFlush,
+                config.getBufferFlushIntervalMs(),
+                config.getBufferFlushIntervalMs(),
+                TimeUnit.MILLISECONDS);
+    }
+
+    public void add(RecordWithMeta record) {
+        if (record == null || StringUtils.isBlank(record.getDatabase()) || StringUtils.isBlank(record.getTable())) {
+            return;
+        }
+        synchronized (lock) {
+            buffer.add(record);
+            currentBytes.addAndGet(estimateSize(record));
+
+            if (buffer.size() >= config.getBufferFlushMaxRows() ||
+                    currentBytes.get() >= config.getBufferFlushMaxBytes()) {
+                flush();
+            }
+        }
+    }
+
+    public void tryFlush() {
+        synchronized (lock) {
+            if (!buffer.isEmpty()) {
+                flush();
+            }
+        }
+    }
+
+    public void flushNow() {
+        synchronized (lock) {
+            flush();
+        }
+        scheduler.shutdownNow();
+    }
+
+    private void flush() {
+        if (buffer.isEmpty()) return;
+
+        List<RecordWithMeta> toSend = new ArrayList<>(buffer);
+        buffer.clear();
+        currentBytes.set(0);
+
+        uploaderExecutor.submit(() -> {
+            try {
+                sinkClient.streamLoad(toSend);
+            } catch (Exception e) {
+                log.error("Stream load failed for {}.{}: {}", tableKey.getDatabase(), tableKey.getTable(), e.getMessage(), e);
+            }
+        });
+    }
+
+    private long estimateSize(RecordWithMeta record) {
+        return record.toString().getBytes(StandardCharsets.UTF_8).length;
+    }
+}

+ 26 - 0
job-adq/src/main/java/cn/com/ctop/job/adq/utils/doris/sink/TableKey.java

@@ -0,0 +1,26 @@
+package cn.com.ctop.job.adq.utils.doris.sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.util.Objects;
+
+@Data
+@AllArgsConstructor
+public class TableKey {
+    private String database;
+    private String table;
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(database, table);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (!(obj instanceof TableKey)) return false;
+        TableKey other = (TableKey) obj;
+        return Objects.equals(database, other.database) && Objects.equals(table, other.table);
+    }
+}

+ 10 - 0
job-adq/src/main/resources/application.yml

@@ -134,6 +134,16 @@ spring:
       min-idle: 8
     port: 6379
 
+  doris:
+    fe-host: http://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:8080
+    user: data
+    password: hr@2025..
+    label-prefix: spring-adq
+    sink-threads: 3
+    sink-batch-rows: 50000
+    sink-batch-mb: 20
+    sink-batch-seconds: 20
+
 #mybatis plus 设置
 mybatis-plus:
   mapper-locations: classpath*:org/jeecg/modules/**/xml/*Mapper.xml,classpath*:cn/com/ctop/**/xml/*Mapper.xml