selectdb stream load

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

@@ -9,6 +9,7 @@ import cn.com.ctop.job.kuaishou.data.service.IKuaiShouErrorLogService;
 import cn.com.ctop.job.kuaishou.data.utils.Check;
 import cn.com.ctop.job.kuaishou.data.utils.DateUtils;
 import cn.com.ctop.job.kuaishou.data.utils.HttpUtils;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -26,17 +27,12 @@ import java.util.*;
 @EnableScheduling
 public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+    @Autowired
     private IKuaiShouErrorLogService errorLogService;
     @Autowired
     private KuaiShouAccountReportMapper accountReportMapper;
 
-   /* private static final int MAX_BATCH_SIZE = 5000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
-
-    private final List<KuaishouAccountReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<KuaishouAccountReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
-    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();*/
 
 
     @Override
@@ -81,12 +77,13 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
                 KuaishouAccountReportHour accountReportHour = JSONObject.toJavaObject(detailJson, KuaishouAccountReportHour.class);
                 accountReportHour.setAdvertiserId(advertiserId);
                 addList.add(accountReportHour);
+
+                dorisLoadManager.add("media_api", "kuaishou_advertiser_report_hourly" , accountReportHour);
             }
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportHour(addList);
-            accountReportMapper.replaceBatcKuaiShouAccountReportHourSD(addList);
-            //advertiserReportHourBuffer.addAll(addList);
+            //accountReportMapper.replaceBatcKuaiShouAccountReportHourSD(addList);
         }
         if (details.size() >= 2000) {
             queryAccountReportHourly(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -136,12 +133,13 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
                 KuaishouAccountReportDaily accountReportDaily = JSONObject.toJavaObject(detailJson, KuaishouAccountReportDaily.class);
                 accountReportDaily.setAdvertiserId(advertiserId);
                 addList.add(accountReportDaily);
+
+                dorisLoadManager.add("media_api", "kuaishou_advertiser_report_daily", accountReportDaily);
             }
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportDaily(addList);
-            accountReportMapper.replaceBatcKuaiShouAccountReportDailySD(addList);
-            //advertiserReportDailyBuffer.addAll(addList);
+            //accountReportMapper.replaceBatcKuaiShouAccountReportDailySD(addList);
         }
         if (details.size() >= 2000) {
             queryAccountReportDaily(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -197,11 +195,13 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
                 KuaishouDailyFlows accountReportDaily = JSONObject.toJavaObject(detailJson, KuaishouDailyFlows.class);
                 accountReportDaily.setAdvertiserId(advertiserId);
                 addList.add(accountReportDaily);
+
+                dorisLoadManager.add("media_api", "kuaishou_advertiser_daily_flows", accountReportDaily);
             }
         }
-        if (!Check.isNull(addList)) {
-            accountReportMapper.replaceKuaiShouDailyFlows(addList);
-        }
+        //if (!Check.isNull(addList)) {
+        //    accountReportMapper.replaceKuaiShouDailyFlows(addList);
+        //}
         if (page < totalPage) {
             queryAccountDailyFlows(accessToken, advertiserId, startDate, endDate, page + 1,isLog, errId, isSend);
         } else {
@@ -221,85 +221,4 @@ public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportS
     public void updateClAccountStatus() {
         accountReportMapper.updateClAccountStatus();
     }
-
-    /*@Scheduled(fixedRate = 14000) // 每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.replaceBatcKuaiShouAccountReportHourSD(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.replaceBatcKuaiShouAccountReportDailySD(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.replaceBatcKuaiShouAccountReportHourSD(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.replaceBatcKuaiShouAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
-                    advertiserReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-    }*/
 }

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

@@ -9,6 +9,7 @@ import cn.com.ctop.job.kuaishou.data.service.IKuaiShouErrorLogService;
 import cn.com.ctop.job.kuaishou.data.utils.Check;
 import cn.com.ctop.job.kuaishou.data.utils.DateUtils;
 import cn.com.ctop.job.kuaishou.data.utils.HttpUtils;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -28,17 +29,11 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
     @Autowired
     private KuaiShouGroupReportMapper kuaiShouGroupReportMapper;
 
-
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
 
-    /*private static final int MAX_BATCH_SIZE = 5000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
-
-    private final List<KuaishouAdgroupReportHourly> reportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<KuaishouAdgroupReportDaily> reportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long reportHourLastFlushTime = System.currentTimeMillis();
-    private long reportDailyLastFlushTime = System.currentTimeMillis();*/
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
 
 
@@ -87,18 +82,13 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
                 KuaishouAdgroupReportHourly dailyGroup = JSONObject.toJavaObject(detailJson, KuaishouAdgroupReportHourly.class);
                 dailyGroup.setAdvertiserId(advertiserId);
                 addList.add(dailyGroup);
+
+                dorisLoadManager.add("media_api", "kuaishou_adgroup_report_hourly", dailyGroup);
             }
         }
         if (!Check.isNull(addList)) {
-            log.info("replaceBatchGroupReportHourly size:{}", addList.size());
-            long t1 = System.currentTimeMillis();
             kuaiShouGroupReportMapper.replaceBatchGroupReportHourly(addList);
-            log.info("replaceBatchGroupReportHourly tidb cost:{}ms", System.currentTimeMillis() - t1);
-            long t2 = System.currentTimeMillis();
-            kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(addList);
-            log.info("replaceBatchGroupReportHourly selectdb cost:{}ms", System.currentTimeMillis() - t2);
-
-            //reportHourBuffer.addAll(addList);
+            //kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(addList);
         }
         if (details.size() >= 2000) {
             queryGroupReportHourly(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -163,12 +153,13 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
                 KuaishouAdgroupReportDaily dailyGroup = JSONObject.toJavaObject(detailJson, KuaishouAdgroupReportDaily.class);
                 dailyGroup.setAdvertiserId(advertiserId);
                 addList.add(dailyGroup);
+
+                dorisLoadManager.add("media_api", "kuaishou_adgroup_report_daily", dailyGroup);
             }
         }
         if (!Check.isNull(addList)) {
             kuaiShouGroupReportMapper.replaceBatchGroupReportDaily(addList);
-            kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(addList);
-            //reportDailyBuffer.addAll(addList);
+            //kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(addList);
         }
         if (details.size() >= 2000) {
             queryGroupReportDaily(accessToken, advertiserId, startDate, endDate, page + 1);
@@ -177,83 +168,4 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
         }
     }
 
-    /*@Scheduled(fixedRate = 13000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行创意报表定时批处理任务....");
-        synchronized (reportHourBuffer) {
-            long now = System.currentTimeMillis();
-            // reportHourBuffer.size() >= MAX_BATCH_SIZE ||  (now - reportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !reportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(new ArrayList<>(reportHourBuffer));
-                    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: {}，执行时间：{}", reportHourBuffer.size(),seconds);
-                    reportHourBuffer.clear();
-                    reportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (reportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            //reportDailyBuffer.size() >= MAX_BATCH_SIZE ||   (now - reportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !reportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(new ArrayList<>(reportDailyBuffer));
-                    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: {}，执行时间：{}", reportDailyBuffer.size(),seconds);
-                    reportDailyBuffer.clear();
-                    reportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启，开始执行创意报表批处理任务....");
-        synchronized (reportHourBuffer) {
-            if (!reportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportHour batch size: {}", reportHourBuffer.size());
-                    kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(new ArrayList<>(reportHourBuffer));
-                    reportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (reportDailyBuffer) {
-            if (!reportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed creativeReportDaily batch size: {}", reportDailyBuffer.size());
-                    kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(new ArrayList<>(reportDailyBuffer));
-                    reportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing creativeReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
 }

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

@@ -14,6 +14,7 @@ import cn.com.ctop.job.kuaishou.data.utils.AlertUtils;
 import cn.com.ctop.job.kuaishou.data.utils.Check;
 import cn.com.ctop.job.kuaishou.data.utils.DateUtils;
 import cn.com.ctop.job.kuaishou.data.utils.HttpUtils;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -29,6 +30,8 @@ import java.util.Map;
 @Slf4j
 public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+    @Autowired
     private KuaiShouRetryReportMapper kuaiShouRetryReportMapper;
     @Autowired
     private KuaiShouAccountReportMapper accountReportMapper;
@@ -88,15 +91,14 @@ public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportServi
                 KuaishouAccountReportHour accountReportHour = JSONObject.toJavaObject(detailJson, KuaishouAccountReportHour.class);
                 accountReportHour.setAdvertiserId(advertiserId);
                 addList.add(accountReportHour);
+
+                dorisLoadManager.add("media_api", "kuaishou_advertiser_report_hourly" , accountReportHour);
+
             }
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportHour(addList);
-            try {
-                accountReportMapper.replaceBatcKuaiShouAccountReportHourSD(addList);
-            }catch (Exception e) {
-                log.error("",e);
-            }
+            //accountReportMapper.replaceBatcKuaiShouAccountReportHourSD(addList);
         }
         if (details.size() >= 2000) {
             getAccountHourRetry(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -149,15 +151,13 @@ public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportServi
                 KuaishouAccountReportDaily accountReportDaily = JSONObject.toJavaObject(detailJson, KuaishouAccountReportDaily.class);
                 accountReportDaily.setAdvertiserId(advertiserId);
                 addList.add(accountReportDaily);
+
+                dorisLoadManager.add("media_api", "kuaishou_advertiser_report_daily", accountReportDaily);
             }
         }
         if (!Check.isNull(addList)) {
             accountReportMapper.replaceBatcKuaiShouAccountReportDaily(addList);
-            try {
-                accountReportMapper.replaceBatcKuaiShouAccountReportDailySD(addList);
-            }catch (Exception e) {
-                log.error("replaceBatcKuaiShouAccountReportDailySD",e);
-            }
+            //accountReportMapper.replaceBatcKuaiShouAccountReportDailySD(addList);
         }
         if (details.size() >= 2000) {
             getAccountDailyRetry(id, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -214,15 +214,13 @@ public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportServi
                 KuaishouAdgroupReportHourly dailyGroup = JSONObject.toJavaObject(detailJson, KuaishouAdgroupReportHourly.class);
                 dailyGroup.setAdvertiserId(advertiserId);
                 addList.add(dailyGroup);
+
+                dorisLoadManager.add("media_api", "kuaishou_adgroup_report_hourly", dailyGroup);
             }
         }
         if (!Check.isNull(addList)) {
             kuaiShouGroupReportMapper.replaceBatchGroupReportHourly(addList);
-            try {
-                kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(addList);
-            }catch (Exception e) {
-                log.error("replaceBatchGroupReportHourlySD",e);
-            }
+            //kuaiShouGroupReportMapper.replaceBatchGroupReportHourlySD(addList);
         }
         if (details.size() >= 2000) {
             getGroupHourRetry(id, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
@@ -279,15 +277,13 @@ public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportServi
                 KuaishouAdgroupReportDaily dailyGroup = JSONObject.toJavaObject(detailJson, KuaishouAdgroupReportDaily.class);
                 dailyGroup.setAdvertiserId(advertiserId);
                 addList.add(dailyGroup);
+
+                dorisLoadManager.add("media_api", "kuaishou_adgroup_report_daily", dailyGroup);
             }
         }
         if (!Check.isNull(addList)) {
             kuaiShouGroupReportMapper.replaceBatchGroupReportDaily(addList);
-            try {
-                kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(addList);
-            }catch (Exception e) {
-                log.error("replaceBatchGroupReportDailySD",e);
-            }
+            //kuaiShouGroupReportMapper.replaceBatchGroupReportDailySD(addList);
         }
         if (details.size() >= 2000) {
             getGroupDailyRetry(id, advertiserId, accessToken, startDate, endDate, page + 1, isSend);

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

@@ -0,0 +1,56 @@
+package cn.com.ctop.job.kuaishou.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);
+    }
+}

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

@@ -0,0 +1,49 @@
+package cn.com.ctop.job.kuaishou.utils;
+
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisBufferConfig;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisMultiTableStreamLoadManager;
+import cn.com.ctop.job.kuaishou.utils.doris.sink.DorisSinkClient;
+import cn.com.ctop.job.kuaishou.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);
+    }
+}

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

@@ -0,0 +1,16 @@
+package cn.com.ctop.job.kuaishou.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;
+}

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

@@ -0,0 +1,51 @@
+package cn.com.ctop.job.kuaishou.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();
+    }
+}

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

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

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

@@ -0,0 +1,17 @@
+package cn.com.ctop.job.kuaishou.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;
+}

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

@@ -0,0 +1,86 @@
+package cn.com.ctop.job.kuaishou.utils.doris.sink;
+
+import cn.com.ctop.job.kuaishou.data.utils.JsonUtils;
+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<Object> dataList = records.stream()
+                .map(RecordWithMeta::getData)
+                .collect(Collectors.toList());
+
+        String jsonBody = JsonUtils.toFilterJSON(dataList);
+        //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);
+        }
+    }
+}

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

@@ -0,0 +1,38 @@
+package cn.com.ctop.job.kuaishou.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);
+    }
+}

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

@@ -0,0 +1,96 @@
+package cn.com.ctop.job.kuaishou.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;
+    }
+}

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

@@ -0,0 +1,26 @@
+package cn.com.ctop.job.kuaishou.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);
+    }
+}

job-kuaishou/src/main/resources/application.yml

@@ -123,7 +123,15 @@ spring:
         # #连接池最大阻塞等待时间（使用负值表示没有限制）
         max-wait: -1ms
 
-
+  doris:
+    fe-host: http://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:8080
+    user: data
+    password: hr@2025..
+    label-prefix: spring-kuaishou
+    sink-threads: 3
+    sink-batch-rows: 50000
+    sink-batch-mb: 20
+    sink-batch-seconds: 20
 
 #mybatis plus 设置
 mybatis-plus: