去除多数据源配置

job-kuaishou/pom.xml

@@ -89,12 +89,6 @@
             <artifactId>druid-spring-boot-starter</artifactId>
             <version>${druid.version}</version>
         </dependency>
-        <!-- 动态数据源 -->
-        <dependency>
-            <groupId>com.baomidou</groupId>
-            <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
-            <version>2.5.4</version>
-        </dependency>
         <!-- junit -->
         <dependency>
             <groupId>junit</groupId>

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/mapper/KuaiShouAccountReportMapper.java

@@ -4,28 +4,21 @@ import cn.com.ctop.job.kuaishou.data.entity.KuaishouAccountReportDaily;
 import cn.com.ctop.job.kuaishou.data.entity.KuaishouAccountReportHour;
 import cn.com.ctop.job.kuaishou.data.entity.KuaishouDailyFlows;
 import com.alibaba.fastjson.JSONObject;
-import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
 public interface KuaiShouAccountReportMapper {
     void replaceBatcKuaiShouAccountReportHour(@Param("adds") List<KuaishouAccountReportHour> addList);
-    @DS("sd_group_commit")
-    void replaceBatcKuaiShouAccountReportHourSD(@Param("adds") List<KuaishouAccountReportHour> addList);
 
     void replaceBatcKuaiShouAccountReportDaily(@Param("adds") List<KuaishouAccountReportDaily> addList);
-    @DS("sd_group_commit")
-    void replaceBatcKuaiShouAccountReportDailySD(@Param("adds") List<KuaishouAccountReportDaily> addList);
 
     void updateKsAccountStatus();
 
     void updateClAccountStatus();
 
-    @DS("sd_group_commit")
     void replaceKuaiShouDailyFlows(List<KuaishouDailyFlows> addList);
 
-    @DS("selectdb")
     List<JSONObject> queryCostAccount(@Param("startDate")Integer startDate,@Param("endDate")Integer endDate);
 
     String getTokenByAdvertiserId(@Param("advertiserId") Long advertiserId);

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/mapper/KuaiShouAdSocialMapper.java

@@ -2,7 +2,6 @@ package cn.com.ctop.job.kuaishou.data.mapper;
 
 import cn.com.ctop.job.kuaishou.data.entity.*;
 import com.alibaba.fastjson.JSONObject;
-import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
@@ -10,47 +9,36 @@ import java.util.List;
 public interface KuaiShouAdSocialMapper {
     void insertTaskList(List<KuaiShouAdSocialTask> list);
 
-    @DS("selectdb")
     void insertTaskListSD(List<KuaiShouAdSocialTask> list);
 
     void insertOrderList(List<KuaiShouAdSocialOrder> list);
 
-    @DS("selectdb")
     void insertOrderListSD(List<KuaiShouAdSocialOrder> list);
 
     void insertOrderDailyData(List<KuaiShouAdSocialOrderDailyData> list);
 
-    @DS("selectdb")
     void insertOrderDailyDataSD(List<KuaiShouAdSocialOrderDailyData> list);
 
     List<JSONObject> queryAdSocialAdvertiserToken(@Param("advertiserId") Long advertiserId);
 
-    @DS("selectdb")
     List<JSONObject> querySavedTask(@Param("advertiserId") Long advertiserId, @Param("endDate") String endDate);
 
-    @DS("selectdb")
     List<JSONObject> querySavedOrder(@Param("advertiserId") Long advertiserId, @Param("endDate") String endDate);
 
     List<JSONObject> queryAdSocialErrorLogs();
 
     List<JSONObject> queryAdSocialContractorToken();
 
-    @DS("selectdb")
     List<JSONObject> queryAdSocialMissionToken(@Param("missionId") Long missionId,@Param("contractorUserId") Long contractorUserId);
 
-    @DS("selectdb")
     void insertContractorMissionList(List<KuaiShouAdSocialContractorMission> addList);
 
-    @DS("selectdb")
     void insertContractorMissionInfo(List<KuaiShouAdSocialContractorMission> addList);
 
-    @DS("selectdb")
     void insertContractorMissionOrder(List<KuaiShouAdSocialContractorMissionOrder> addList);
 
-    @DS("selectdb")
     List<JSONObject> test1();
 
-    @DS("sd_group_commit")
     List<JSONObject> test2();
 
 }

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/mapper/KuaiShouDramaMapper.java

@@ -7,7 +7,6 @@ import cn.com.ctop.job.kuaishou.data.entity.KuaiShouDramaOrder;
 import cn.com.ctop.job.kuaishou.data.entity.KuaiShouDramaSalerCopyRightData;
 import cn.com.ctop.job.kuaishou.data.entity.KuaiShouDramaSeriesInfo;
 import com.alibaba.fastjson.JSONObject;
-import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
@@ -15,15 +14,15 @@ import java.util.Set;
 
 public interface KuaiShouDramaMapper {
     void insertAdReportDetail(List<KuaiShouDramaAdReportDetail> adReportDetails);
-    @DS("selectdb")
+
     void insertAdReportDetailSD(List<KuaiShouDramaAdReportDetail> adReportDetails);
 
     void insertDramaOrder(List<KuaiShouDramaOrder> orderList);
-    @DS("selectdb")
+
     void insertDramaOrderSD(List<KuaiShouDramaOrder> orderList);
 
     void insertSalerCopyRightData(List<KuaiShouDramaSalerCopyRightData> salerCopyRightDataList);
-    @DS("selectdb")
+
     void insertSalerCopyRightDataSD(List<KuaiShouDramaSalerCopyRightData> salerCopyRightDataList);
 
 

job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/data/mapper/KuaiShouGroupReportMapper.java

@@ -3,7 +3,6 @@ package cn.com.ctop.job.kuaishou.data.mapper;
 import cn.com.ctop.job.kuaishou.data.entity.KuaishouAdgroupReportDaily;
 import cn.com.ctop.job.kuaishou.data.entity.KuaishouAdgroupReportHourly;
 import com.alibaba.fastjson.JSONObject;
-import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
@@ -11,11 +10,11 @@ import java.util.List;
 public interface KuaiShouGroupReportMapper {
 
     void replaceBatchGroupReportHourly(@Param("addList") List<KuaishouAdgroupReportHourly> addList);
-    @DS("sd_group_commit")
+
     void replaceBatchGroupReportHourlySD(@Param("addList") List<KuaishouAdgroupReportHourly> addList);
 
     void replaceBatchGroupReportDaily(@Param("addList") List<KuaishouAdgroupReportDaily> addList);
-    @DS("sd_group_commit")
+
     void replaceBatchGroupReportDailySD(@Param("addList") List<KuaishouAdgroupReportDaily> addList);
 
     List<Long> getAdvertiserIdList(@Param("type") Integer type);

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

@@ -9,7 +9,6 @@ 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;
@@ -27,8 +26,6 @@ import java.util.*;
 @EnableScheduling
 public class KuaiShouAccountReportServiceImpl implements IKuaiShouAccountReportService {
     @Autowired
-    private DorisMultiTableStreamLoadManager dorisLoadManager;
-    @Autowired
     private IKuaiShouErrorLogService errorLogService;
     @Autowired
     private KuaiShouAccountReportMapper accountReportMapper;
@@ -77,8 +74,6 @@ 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)) {
@@ -133,8 +128,6 @@ 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)) {
@@ -203,8 +196,6 @@ 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)) {

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

@@ -9,7 +9,6 @@ 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;
@@ -22,8 +21,6 @@ import java.util.*;
 @Service
 public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
     @Autowired
-    private DorisMultiTableStreamLoadManager dorisLoadManager;
-    @Autowired
     private KuaiShouAdSocialMapper kuaiShouAdSocialMapper;
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
@@ -76,7 +73,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertTaskList(addList);
             //kuaiShouAdSocialMapper.insertTaskListSD(addList);
-            dorisLoadManager.add("media_api", "kuaishou_ad_social_task", addList);
         }
         if (page < totalPage) {
             getTaskList(accessToken, advertiserId, startDate, endDate, page + 1, isLog, errId, isSend);
@@ -144,7 +140,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertOrderList(addList);
             //kuaiShouAdSocialMapper.insertOrderListSD(addList);
-            dorisLoadManager.add("media_api", "kuaishou_ad_social_task_order", addList);
         }
         if (page < totalPage) {
             getOrderList(accessToken, advertiserId, taskId, startDate, endDate, page + 1, isLog, errId, isSend);
@@ -229,7 +224,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         if (!Check.isNull(addList)) {
             kuaiShouAdSocialMapper.insertOrderDailyData(addList);
             //kuaiShouAdSocialMapper.insertOrderDailyDataSD(addList);
-            dorisLoadManager.add("media_api", "kuaishou_ad_social_task_order_report_daily", addList);
         }
 
         log.info("【{} {} {}】录入快手星任务（视频）-订单分天数据完成 {} {}", advertiserId, taskId, orderIds, startDate, endDate);
@@ -309,10 +303,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
                 addList.add(mission);
             }
         }
-        if (!Check.isNull(addList)) {
-            //kuaiShouAdSocialMapper.insertContractorMissionList(addList);
-            dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission", addList, "mission_id,contractor_user_id,advertiser_id,status");
-        }
         if (offset + pageSize < totalCount) {
             getContractorMissionList(accessToken, advertiserId, contractorId, offset + pageSize);
         } else {
@@ -359,9 +349,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
         mission.setAdvertiserId(advertiserId);
         mission.setContractorUserId(contractorId);
         //kuaiShouAdSocialMapper.insertContractorMissionInfo(Collections.singletonList(mission));
-        dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission", mission,
-                "mission_id,contractor_user_id,advertiser_id, mission_name, mission_requirement," +
-                        "logo_url, uint_price_type, contractor_commission_ratio, uint_price_amount, progress_rate, budget,start_time,end_time");
         log.info("【{}】获取快手团长任务详情完成 {} {}", advertiserId, contractorId, missionId);
     }
 
@@ -419,10 +406,6 @@ public class KuaiShouAdSocialServiceImpl implements IKuaiShouAdSocialService {
                 addList.add(order);
             }
         }
-        if (!Check.isNull(addList)) {
-            //kuaiShouAdSocialMapper.insertContractorMissionOrder(addList);
-            dorisLoadManager.add("media_api", "kuaishou_ad_social_transaction_contractor_mission_order", addList);
-        }
         if (offset + pageSize < totalCount && offset + pageSize <= 10000 - pageSize) {
             getContractorMissionOrder(accessToken, advertiserId, contractorId, missionId, filterStartTime, offset + pageSize , 0);
         } else if (offset + pageSize < totalCount && offset + pageSize > 10000 - pageSize) {

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

@@ -14,7 +14,6 @@ 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;
@@ -30,8 +29,6 @@ public class KuaiShouDramaServiceImpl implements IKuaiShouDramaService {
     private KuaiShouDramaMapper kuaiShouDramaMapper;
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
-    @Autowired
-    private DorisMultiTableStreamLoadManager dorisStreamLoadManager;
 
     @Override
     public void queryAdReportDetail(String accessToken, Long advertiserId, String startDate, String endDate, int page, Integer isLog, Long errId, Integer isSend) {
@@ -216,12 +213,8 @@ public class KuaiShouDramaServiceImpl implements IKuaiShouDramaService {
         if (!Check.isNull(details)) {
             for (int i = 0; i < details.size(); i++) {
                 JSONObject detailJson = details.getJSONObject(i);
-                detailJson.put("order_type", 2);
-                detailJson.put("stat_date", DateUtils.stampToIntDate(detailJson.getLong("create_time")));
-                detailJson.put("admin_advertiser_id", advertiserId);
-                detailJson.put("series_name", getSeriesName(advertiserId, detailJson.getLong("series_id"), accessToken));
+                getSeriesName(advertiserId, detailJson.getLong("series_id"), accessToken);
             }
-            dorisStreamLoadManager.add("media_api", "kuaishou_drama_order", details);
         }
 
         if (page < totalPage) {
@@ -348,16 +341,6 @@ public class KuaiShouDramaServiceImpl implements IKuaiShouDramaService {
         int totalCount = data.getIntValue("total_count");
         int totalPage = (int) Math.ceil(totalCount * 1.0 / pageSize);
         JSONArray details = data.getJSONArray("data_list");
-        if (!Check.isNull(details)) {
-            for (int i = 0; i < details.size(); i++) {
-                JSONObject detailJson = details.getJSONObject(i);
-                detailJson.put("stat_date", DateUtils.getDateInteger(detailJson.getString("report_date")));
-                detailJson.put("admin_advertiser_id", advertiserId);
-                detailJson.put("series_type", seriesType);
-            }
-            dorisStreamLoadManager.add("media_api", "kuaishou_drama_saler_data", details);
-        }
-
         if (page < totalPage) {
             querySalerData(accessToken, advertiserId, seriesType, startDate, endDate, page + 1, isLog, errId, isSend);
         } else {

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

@@ -10,7 +10,6 @@ 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;
@@ -33,9 +32,6 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
     @Autowired
     private IKuaiShouErrorLogService errorLogService;
 
-    @Autowired
-    private DorisMultiTableStreamLoadManager dorisLoadManager;
-
 
     @Override
     public void queryGroupReportHourly(String accessToken, Long advertiserId, String startDate, String endDate, int page) {
@@ -82,8 +78,6 @@ 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)) {
@@ -153,8 +147,6 @@ 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)) {
@@ -200,23 +192,6 @@ public class KuaiShouGroupReportServiceImpl implements IKuaiShouGroupReportServi
             return;
         }
         JSONArray details = resultJson.getJSONObject("data").getJSONArray("details");
-        if (!Check.isNull(details)) {
-            for (int i = 0; i < details.size(); i++) {
-                JSONObject detailJson = details.getJSONObject(i);
-                JSONObject data = new JSONObject();
-                data.put("advertiser_id", advertiserId);
-                data.put("unit_id", detailJson.getLong("unit_id"));
-                data.put("campaign_id", detailJson.getString("campaign_id"));
-                data.put("series_id", detailJson.getString("series_id"));
-                data.put("episode_id", detailJson.getString("episode_id"));
-                data.put("create_time", detailJson.getString("create_time"));
-                data.put("update_time", detailJson.getString("update_time"));
-                data.put("study_status", detailJson.getString("study_status"));
-                data.put("status", detailJson.getString("status"));
-                data.put("put_status", detailJson.getString("put_status"));
-                dorisLoadManager.add("media_api", "kuaishou_adgroup_list", data);
-            }
-        }
 
         if (details.size() >= pageSize) {
             queryGroupList(accessToken, advertiserId, startDate, endDate, page + 1, isLog, errId, isSend);

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

@@ -14,7 +14,6 @@ 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;
@@ -30,8 +29,6 @@ import java.util.Map;
 @Slf4j
 public class KuaiShouRetryReportServiceImpl implements IKuaiShouRetryReportService {
     @Autowired
-    private DorisMultiTableStreamLoadManager dorisLoadManager;
-    @Autowired
     private KuaiShouRetryReportMapper kuaiShouRetryReportMapper;
     @Autowired
     private KuaiShouAccountReportMapper accountReportMapper;
@@ -91,9 +88,6 @@ 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)) {
@@ -151,8 +145,6 @@ 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)) {
@@ -214,8 +206,6 @@ 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)) {
@@ -277,8 +267,6 @@ 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)) {

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

@@ -1,49 +0,0 @@
-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

@@ -1,16 +0,0 @@
-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

@@ -1,65 +0,0 @@
-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));
-        }
-    }
-
-    public void add(String database, String table, Object obj, String updateColumns) {
-        TableKey key = new TableKey(database, table,updateColumns);
-        if (obj instanceof Collection) {
-            Collection<?> collection = (Collection<?>) obj;
-            for (Object o : collection) {
-                bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
-                        .add(new RecordWithMeta(database, table, o, updateColumns));
-            }
-        } else {
-            bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
-                    .add(new RecordWithMeta(database, table, obj, updateColumns));
-        }
-    }
-
-
-    @PreDestroy
-    public void shutdown() {
-        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

@@ -1,7 +0,0 @@
-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

@@ -1,24 +0,0 @@
-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;
-    private String updateColumns;
-
-    public RecordWithMeta(String database, String table, Object data) {
-        this.database = database;
-        this.table = table;
-        this.data = data;
-    }
-}

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

@@ -1,99 +0,0 @@
-package cn.com.ctop.job.kuaishou.utils.doris.sink;
-
-import cn.com.ctop.job.kuaishou.data.utils.JsonUtils;
-import com.alibaba.fastjson.JSONObject;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.HttpEntity;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.client.methods.HttpPut;
-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 updateColumns = sample.getUpdateColumns();
-        String label = labelPrefix + "_" + db + "_" + table + "_" + UUID.randomUUID();
-        String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
-
-        // 提取 data
-        List<String> dataList = records.stream()
-                .map(d -> {
-                    Object data = d.getData();
-                    if (data instanceof JSONObject) {
-                        return data.toString();
-                    } else {
-                        return JsonUtils.toFilterJSONString(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);
-        }
-        if (StringUtils.isNotBlank(updateColumns)) {
-            put.setHeader("partial_columns", "true");
-            put.setHeader("columns", updateColumns);
-        }
-
-        put.setHeader("format", format); // json
-        put.setHeader("strip_outer_array", "true"); // json array
-        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

@@ -1,38 +0,0 @@
-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

@@ -1,96 +0,0 @@
-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

@@ -1,27 +0,0 @@
-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;
-    private String updateColumns;
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(database, table, updateColumns);
-    }
-
-    @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) && Objects.equals(updateColumns, other.updateColumns);
-    }
-}

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

@@ -44,61 +44,36 @@ spring:
     static-path-pattern: /**
   resource:
     static-locations: classpath:/static/,classpath:/public/
-  autoconfigure:
-    exclude: com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure
   datasource:
-    #    name: fendo
-    #    url: jdbc:mysql://39.97.120.42:3306/track?characterEncoding=UTF-8&useUnicode=true&useSSL=false
-    #    username: hcst
-    #    password: test@20190531
+    url: jdbc:mysql://172.16.0.117:3390/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false
+    username: data
+    password: hr@2025..
     type: com.alibaba.druid.pool.DruidDataSource
     driver-class-name: com.mysql.jdbc.Driver
-    dynamic:
-      primary: master  # 设置默认的数据源
-      strict: false    # 严格匹配数据源，未匹配到时使用默认数据源
-      druid: # 全局druid参数，绝大部分值和默认保持一致。(现已支持的参数如下,不清楚含义不要乱设置)
-        # 连接池的配置信息
-        # 初始化大小，最小，最大
-        connection-init-sql: set names utf8mb4
-
-        initial-size: 5
-        min-idle: 5
-        maxActive: 20
-        # 配置获取连接等待超时的时间
-        maxWait: 60000
-        # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
-        timeBetweenEvictionRunsMillis: 60000
-        # 配置一个连接在池中最小生存的时间，单位是毫秒
-        minEvictableIdleTimeMillis: 300000
-        validationQuery: SELECT 1 FROM DUAL
-        testWhileIdle: true
-        testOnBorrow: false
-        testOnReturn: false
-        # 打开PSCache，并且指定每个连接上PSCache的大小
-        poolPreparedStatements: true
-        maxPoolPreparedStatementPerConnectionSize: 20
-        # 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
-        filters: stat,slf4j
-        # 通过connectProperties属性来打开mergeSql功能；慢SQL记录
-        connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
-      datasource:
-        master:
-          url: jdbc:mysql://172.16.0.117:3390/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false
-         # url: jdbc:mysql://8.147.135.80:3390/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false
-          username: data
-          password: hr@2025..
-          driver-class-name: com.mysql.jdbc.Driver
-        selectdb:
-          url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false&sessionVariables=enable_unique_key_partial_update=true,enable_insert_strict=false
-          username: data
-          password: hr@2025..
-          driver-class-name: com.mysql.jdbc.Driver
-        sd_group_commit:
-          ## 注意此selectdb数据源不支持列更新，insert into 对于未指定列会被置为null，类似tidb replace into
-          url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false&useServerPrepStmts=true&useLocalSessionState=true&rewriteBatchedStatements=true&cachePrepStmts=true&prepStmtCacheSqlLimit=99999&prepStmtCacheSize=500&sessionVariables=group_commit=async_mode,enable_nereids_planner=false
-          username: data
-          password: hr@2025..
-          driver-class-name: com.mysql.jdbc.Driver
+    druid:
+      # 连接池的配置信息
+      # 初始化大小，最小，最大
+      connection-init-sql: set names utf8mb4
+      initial-size: 5
+      min-idle: 5
+      maxActive: 20
+      # 配置获取连接等待超时的时间
+      maxWait: 60000
+      # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
+      timeBetweenEvictionRunsMillis: 60000
+      # 配置一个连接在池中最小生存的时间，单位是毫秒
+      minEvictableIdleTimeMillis: 300000
+      validationQuery: SELECT 1 FROM DUAL
+      testWhileIdle: true
+      testOnBorrow: false
+      testOnReturn: false
+      # 打开PSCache，并且指定每个连接上PSCache的大小
+      poolPreparedStatements: true
+      maxPoolPreparedStatementPerConnectionSize: 20
+      # 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
+      filters: stat,slf4j
+      # 通过connectProperties属性来打开mergeSql功能；慢SQL记录
+      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
 
   # redis 配置
   redis:
@@ -123,16 +98,6 @@ 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:
   mapper-locations: classpath*:org/jeecg/modules/**/xml/*Mapper.xml,classpath*:cn/com/ctop/**/xml/*Mapper.xml