每 3小时回滚的 bug

flink-ad-monitoring/src/main/java/flink/zanxiangnet/ad/monitoring/process/AdHourDWDProcess.java

@@ -72,7 +72,6 @@ public class AdHourDWDProcess extends KeyedProcessFunction<Long, AdDataOfHourODS
     public void processElement(AdDataOfHourODS element, KeyedProcessFunction<Long, AdDataOfHourODS, AdStatOfHourDWD>.Context context, Collector<AdStatOfHourDWD> collector) throws Exception {
         LocalDate statDay = DateUtil.parseLocalDate(element.getStatDay());
         long now = System.currentTimeMillis();
-        LocalDate today = LocalDate.now();
         long adId = element.getAdId();
 
         // 每小时从 ck更新下历史的天数据
@@ -101,7 +100,11 @@ public class AdHourDWDProcess extends KeyedProcessFunction<Long, AdDataOfHourODS
             dayODSHourMap = new HashMap<>(24);
         }
         if (!dayODSHourMap.isEmpty()) {
-            dayODSList = new ArrayList<>(dayODSHourMap.values());
+            dayODSHourMap.values().forEach(hourODS -> {
+                if (hourODS.getHour() < element.getHour()) {
+                    dayODSList.add(hourODS);
+                }
+            });
         }
 
         AdStatOfHourDWD newStatData = AdStatOfHourDWD.reduce(yesterdayReduceData, dayODSList, element, now);

flink-ad-monitoring/src/main/java/flink/zanxiangnet/ad/monitoring/process/CostHourProcess.java

@@ -106,85 +106,85 @@ public class CostHourProcess extends KeyedProcessFunction<Long, AdStatOfHourDWD,
     }
 
     //数据格式转换
-    public CostHourDM dataChange(AdStatOfHourDWD adStatOfMinuteDWD) {
+    public CostHourDM dataChange(AdStatOfHourDWD hourDWD) {
         CostHourDM costHourDM = new CostHourDM();
 
         //时间-天
-        costHourDM.setDt(adStatOfMinuteDWD.getStatDay());
+        costHourDM.setDt(hourDWD.getStatDay());
         //计划 id
-        costHourDM.setCampaignId(adStatOfMinuteDWD.getCampaignId().toString());
+        costHourDM.setCampaignId(hourDWD.getCampaignId().toString());
         //时间- real
-        costHourDM.setCreateTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(adStatOfMinuteDWD.getCreateTime()));
+        costHourDM.setCreateTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(hourDWD.getCreateTime()));
         //时间-小时
         //TODO:之后需要进一步修改
-        String tmpHour = adStatOfMinuteDWD.getHour() > 9 ? adStatOfMinuteDWD.getHour().toString() : "0" + adStatOfMinuteDWD.getHour().toString();
-        costHourDM.setHour(adStatOfMinuteDWD.getStatDay() + " " + tmpHour + ":00:00");
+        String tmpHour = hourDWD.getHour() > 9 ? hourDWD.getHour().toString() : "0" + hourDWD.getHour().toString();
+        costHourDM.setHour(hourDWD.getStatDay() + " " + tmpHour + ":00:00");
         //广告id
-        costHourDM.setAdId(adStatOfMinuteDWD.getAdId().toString());
+        costHourDM.setAdId(hourDWD.getAdId().toString());
         //服务商id
-        costHourDM.setAgencyAccountId(adStatOfMinuteDWD.getAgencyAccountId().toString());
+        costHourDM.setAgencyAccountId(hourDWD.getAgencyAccountId().toString());
         //广告组id
-        costHourDM.setAdgroupId(adStatOfMinuteDWD.getAdgroupId().toString());
+        costHourDM.setAdgroupId(hourDWD.getAdgroupId().toString());
         //创意id
         costHourDM.setAdcreativeId("");
         //账号id
-        costHourDM.setAccountId(adStatOfMinuteDWD.getAccountId().toString());
+        costHourDM.setAccountId(hourDWD.getAccountId().toString());
         //总消耗
-        costHourDM.setCostTotal(adStatOfMinuteDWD.getCostTotal());
+        costHourDM.setCostTotal(hourDWD.getCostTotal());
         //当天消耗
-        costHourDM.setCostDay(adStatOfMinuteDWD.getCostDay());
+        costHourDM.setCostDay(hourDWD.getCostDay());
         //当天小时消耗
-        costHourDM.setCostHour(adStatOfMinuteDWD.getCostHour());
+        costHourDM.setCostHour(hourDWD.getCostHour());
         //消耗速度
-        costHourDM.setCostSpeed(adStatOfMinuteDWD.getCostHour());
+        costHourDM.setCostSpeed(hourDWD.getCostHour());
         //总浏览量
-        costHourDM.setViewCountTotal(adStatOfMinuteDWD.getViewCountTotal());
+        costHourDM.setViewCountTotal(hourDWD.getViewCountTotal());
         //天-总浏览量
-        costHourDM.setViewCountDay(adStatOfMinuteDWD.getViewCountDay());
+        costHourDM.setViewCountDay(hourDWD.getViewCountDay());
         //小时-总浏览量
-        costHourDM.setViewCountHour(adStatOfMinuteDWD.getViewCountHour());
+        costHourDM.setViewCountHour(hourDWD.getViewCountHour());
         //总平均千次曝光成本
-        costHourDM.setThousandDisplayPriceAll(adStatOfMinuteDWD.getThousandDisplayPriceAll());
+        costHourDM.setThousandDisplayPriceAll(hourDWD.getThousandDisplayPriceAll());
         //天-总平均曝光成本
-        costHourDM.setThousandDisplayPriceDay(adStatOfMinuteDWD.getThousandDisplayPriceDay());
+        costHourDM.setThousandDisplayPriceDay(hourDWD.getThousandDisplayPriceDay());
         //小时-总平均曝光成本
-        costHourDM.setThousandDisplayPriceHour(adStatOfMinuteDWD.getThousandDisplayPriceHour());
+        costHourDM.setThousandDisplayPriceHour(hourDWD.getThousandDisplayPriceHour());
         //总点击量
-        costHourDM.setValidClickCountTotal(adStatOfMinuteDWD.getValidClickCountTotal());
+        costHourDM.setValidClickCountTotal(hourDWD.getValidClickCountTotal());
         //天-总点击量
-        costHourDM.setValidClickCountDay(adStatOfMinuteDWD.getValidClickCountDay());
+        costHourDM.setValidClickCountDay(hourDWD.getValidClickCountDay());
         //小时-总点击量
-        costHourDM.setValidClickCountHour(adStatOfMinuteDWD.getValidClickCountHour());
+        costHourDM.setValidClickCountHour(hourDWD.getValidClickCountHour());
         //总平均点击率
-        costHourDM.setCtrAll(adStatOfMinuteDWD.getCtrAll().doubleValue());
+        costHourDM.setCtrAll(hourDWD.getCtrAll().doubleValue());
         //天-总平均点击率
-        costHourDM.setCtrDay(adStatOfMinuteDWD.getCtrDay().doubleValue());
+        costHourDM.setCtrDay(hourDWD.getCtrDay().doubleValue());
         //小时-总平均点击率
-        costHourDM.setCtrHour(adStatOfMinuteDWD.getCtrHour().doubleValue());
+        costHourDM.setCtrHour(hourDWD.getCtrHour().doubleValue());
         //总点击均价
-        costHourDM.setCpcAll(adStatOfMinuteDWD.getCpcAll());
+        costHourDM.setCpcAll(hourDWD.getCpcAll());
         //天-总点击均价
-        costHourDM.setCpcDay(adStatOfMinuteDWD.getCpcDay());
+        costHourDM.setCpcDay(hourDWD.getCpcDay());
         //小时-总点击均价
-        costHourDM.setCpcHour(adStatOfMinuteDWD.getCpcHour());
+        costHourDM.setCpcHour(hourDWD.getCpcHour());
         //总目标转化量
-        costHourDM.setConversionsCountTotal(adStatOfMinuteDWD.getConversionsCountTotal());
+        costHourDM.setConversionsCountTotal(hourDWD.getConversionsCountTotal());
         //天-总目标转化量
-        costHourDM.setConversionsCountDay(adStatOfMinuteDWD.getConversionsCountDay());
+        costHourDM.setConversionsCountDay(hourDWD.getConversionsCountDay());
         //小时-总目标转化量
-        costHourDM.setConversionsCountHour(adStatOfMinuteDWD.getConversionsCountHour());
+        costHourDM.setConversionsCountHour(hourDWD.getConversionsCountHour());
         //总目标平均转化成本
-        costHourDM.setConversionsCostTotal(adStatOfMinuteDWD.getConversionsCostAll());
+        costHourDM.setConversionsCostTotal(hourDWD.getConversionsCostAll());
         //天-总目标平均转化成本
-        costHourDM.setConversionsCostDay(adStatOfMinuteDWD.getConversionsCostDay());
+        costHourDM.setConversionsCostDay(hourDWD.getConversionsCostDay());
         //小时-总目标平均转化成本
-        costHourDM.setConversionsCostHour(adStatOfMinuteDWD.getConversionsCostHour());
+        costHourDM.setConversionsCostHour(hourDWD.getConversionsCostHour());
         //总平均转化率
-        costHourDM.setConversionsRateAll(adStatOfMinuteDWD.getConversionsRateAll().doubleValue());
+        costHourDM.setConversionsRateAll(hourDWD.getConversionsRateAll().doubleValue());
         //天-总平均转化率
-        costHourDM.setConversionsRateDay(adStatOfMinuteDWD.getConversionsRateDay().doubleValue());
+        costHourDM.setConversionsRateDay(hourDWD.getConversionsRateDay().doubleValue());
         //小时-总平均转化率
-        costHourDM.setConversionsRateHour(adStatOfMinuteDWD.getConversionsRateHour().doubleValue());
+        costHourDM.setConversionsRateHour(hourDWD.getConversionsRateHour().doubleValue());
         //TODO:总首日下单roi
         costHourDM.setFirstDayOrderRoiTotal(0);
         //天-总首日下单roi
@@ -192,89 +192,89 @@ public class CostHourProcess extends KeyedProcessFunction<Long, AdStatOfHourDWD,
         //小时-总首日下单roi
         costHourDM.setFirstDayOrderRoiHour(0);
         //总首日下单金额
-        costHourDM.setFirstDayOrderAmountTotal(adStatOfMinuteDWD.getFirstDayOrderAmountTotal());
+        costHourDM.setFirstDayOrderAmountTotal(hourDWD.getFirstDayOrderAmountTotal());
         //天-总首日下单金额
-        costHourDM.setFirstDayOrderAmountDay(adStatOfMinuteDWD.getFirstDayOrderAmountDay());
+        costHourDM.setFirstDayOrderAmountDay(hourDWD.getFirstDayOrderAmountDay());
         //小时-总首日下单金额
-        costHourDM.setFirstDayOrderAmountHour(adStatOfMinuteDWD.getFirstDayOrderAmountHour());
+        costHourDM.setFirstDayOrderAmountHour(hourDWD.getFirstDayOrderAmountHour());
         //总首日下单量
-        costHourDM.setFirstDayOrderCountTotal(adStatOfMinuteDWD.getFirstDayOrderCountTotal());
+        costHourDM.setFirstDayOrderCountTotal(hourDWD.getFirstDayOrderCountTotal());
         //天-总首日下单量
-        costHourDM.setFirstDayOrderCountDay(adStatOfMinuteDWD.getFirstDayOrderCountDay());
+        costHourDM.setFirstDayOrderCountDay(hourDWD.getFirstDayOrderCountDay());
         //小时-总首日下单量
-        costHourDM.setFirstDayOrderCountHour(adStatOfMinuteDWD.getFirstDayOrderCountHour());
+        costHourDM.setFirstDayOrderCountHour(hourDWD.getFirstDayOrderCountHour());
         //总下单金额
-        costHourDM.setWebOrderAmountTotal(adStatOfMinuteDWD.getOrderAmountTotal());
+        costHourDM.setWebOrderAmountTotal(hourDWD.getOrderAmountTotal());
         //天-总下单金额
-        costHourDM.setWebOrderAmountDay(adStatOfMinuteDWD.getOrderAmountDay());
+        costHourDM.setWebOrderAmountDay(hourDWD.getOrderAmountDay());
         //小时-总下单金额
-        costHourDM.setWebOrderAmountHour(adStatOfMinuteDWD.getOrderAmountHour());
+        costHourDM.setWebOrderAmountHour(hourDWD.getOrderAmountHour());
         //总平均下单成本
-        costHourDM.setWebOrderCostTotal(adStatOfMinuteDWD.getWebOrderCostAll());
+        costHourDM.setWebOrderCostTotal(hourDWD.getWebOrderCostAll());
         //天-总平均下单成本
-        costHourDM.setWebOrderCostDay(adStatOfMinuteDWD.getWebOrderCostDay());
+        costHourDM.setWebOrderCostDay(hourDWD.getWebOrderCostDay());
         //小时-总平均下单成本
-        costHourDM.setWebOrderCostHour(adStatOfMinuteDWD.getWebOrderCostHour());
+        costHourDM.setWebOrderCostHour(hourDWD.getWebOrderCostHour());
         //总平均下单率
-        costHourDM.setWebOrderRateTotal(adStatOfMinuteDWD.getOrderRateAll().doubleValue());
+        costHourDM.setWebOrderRateTotal(hourDWD.getOrderRateAll().doubleValue());
         //天-总平均下单率
-        costHourDM.setWebOrderRateDay(adStatOfMinuteDWD.getOrderRateDay().doubleValue());
+        costHourDM.setWebOrderRateDay(hourDWD.getOrderRateDay().doubleValue());
         //小时-总平均下单率
-        costHourDM.setWebOrderRateHour(adStatOfMinuteDWD.getOrderRateHour().doubleValue());
+        costHourDM.setWebOrderRateHour(hourDWD.getOrderRateHour().doubleValue());
         //TODO:总平均下单量-----webordercount和ordercount是同一个东西吗
-        costHourDM.setWebOrderCountTotal(adStatOfMinuteDWD.getOrderCountTotal());
+        costHourDM.setWebOrderCountTotal(hourDWD.getOrderCountTotal());
         //天-总平均下单量
-        costHourDM.setWebOrderCountDay(adStatOfMinuteDWD.getOrderCountDay());
+        costHourDM.setWebOrderCountDay(hourDWD.getOrderCountDay());
         //小时-总平均下单量
-        costHourDM.setWebOrderCountHour(adStatOfMinuteDWD.getOrderCountHour());
+        costHourDM.setWebOrderCountHour(hourDWD.getOrderCountHour());
         //总下单ROI
-        costHourDM.setOrderRoiTotal(adStatOfMinuteDWD.getOrderRoiAll().doubleValue());
+        costHourDM.setOrderRoiTotal(hourDWD.getOrderRoiAll().doubleValue());
         //天-总下单roi
-        costHourDM.setOrderRoiDay(adStatOfMinuteDWD.getOrderRoiDay().doubleValue());
+        costHourDM.setOrderRoiDay(hourDWD.getOrderRoiDay().doubleValue());
         //小时-总下单roi
-        costHourDM.setOrderRoiHour(adStatOfMinuteDWD.getOrderRoiHour().doubleValue());
+        costHourDM.setOrderRoiHour(hourDWD.getOrderRoiHour().doubleValue());
         //总平均下单客单价
-        costHourDM.setOrderUnitPriceTotal(adStatOfMinuteDWD.getOrderUnitPriceAll());
+        costHourDM.setOrderUnitPriceTotal(hourDWD.getOrderUnitPriceAll());
         //天-总平均下单客单价
-        costHourDM.setOrderUnitPriceDay(adStatOfMinuteDWD.getOrderUnitPriceDay());
+        costHourDM.setOrderUnitPriceDay(hourDWD.getOrderUnitPriceDay());
         //小时-总平均下单客单价
-        costHourDM.setOrderUnitPriceHour(adStatOfMinuteDWD.getOrderUnitPriceHour());
+        costHourDM.setOrderUnitPriceHour(hourDWD.getOrderUnitPriceHour());
         //总公众号关注量
-        costHourDM.setFromFollowUvTotal(adStatOfMinuteDWD.getOfficialAccountFollowCountTotal());
+        costHourDM.setFromFollowUvTotal(hourDWD.getOfficialAccountFollowCountTotal());
         //天-总公众号关注量
-        costHourDM.setFromFollowUvDay(adStatOfMinuteDWD.getOfficialAccountFollowCountDay());
+        costHourDM.setFromFollowUvDay(hourDWD.getOfficialAccountFollowCountDay());
         //小时-总公众号关注量
-        costHourDM.setFromFollowUvHour(adStatOfMinuteDWD.getOfficialAccountFollowCountHour());
+        costHourDM.setFromFollowUvHour(hourDWD.getOfficialAccountFollowCountHour());
         //TODO:总平均公众号关注成本---是否是价格/关注
-        costHourDM.setFromFollowCostTotal(adStatOfMinuteDWD.getOfficialAccountFollowCountTotal() == 0 ? 0 : adStatOfMinuteDWD.getCostTotal() / adStatOfMinuteDWD.getOfficialAccountFollowCountTotal());
+        costHourDM.setFromFollowCostTotal(hourDWD.getOfficialAccountFollowCountTotal() == 0 ? 0 : hourDWD.getCostTotal() / hourDWD.getOfficialAccountFollowCountTotal());
         //天-总平均公众号关注成本
-        costHourDM.setFromFollowCostDay(adStatOfMinuteDWD.getOfficialAccountFollowCountDay() == 0 ? 0 : adStatOfMinuteDWD.getCostDay() / adStatOfMinuteDWD.getOfficialAccountFollowCountDay());
+        costHourDM.setFromFollowCostDay(hourDWD.getOfficialAccountFollowCountDay() == 0 ? 0 : hourDWD.getCostDay() / hourDWD.getOfficialAccountFollowCountDay());
         //小时-总平均公众号关注成本
-        costHourDM.setFromFollowCostHour(adStatOfMinuteDWD.getOfficialAccountFollowCountHour() == 0 ? 0 : adStatOfMinuteDWD.getCostHour() / adStatOfMinuteDWD.getOfficialAccountFollowCountHour());
+        costHourDM.setFromFollowCostHour(hourDWD.getOfficialAccountFollowCountHour() == 0 ? 0 : hourDWD.getCostHour() / hourDWD.getOfficialAccountFollowCountHour());
         //TODO:总平均公众号关注率----确认是否对应
-        costHourDM.setFromFollowRateTotal(adStatOfMinuteDWD.getOfficialAccountFollowRateAll().doubleValue());
+        costHourDM.setFromFollowRateTotal(hourDWD.getOfficialAccountFollowRateAll().doubleValue());
         //天-总平均公众号关注率
-        costHourDM.setFromFollowRateDay(adStatOfMinuteDWD.getOfficialAccountFollowRateDay().doubleValue());
+        costHourDM.setFromFollowRateDay(hourDWD.getOfficialAccountFollowRateDay().doubleValue());
         //小时-总平均公众号关注率
-        costHourDM.setFromFollowRateHour(adStatOfMinuteDWD.getOfficialAccountFollowRateHour().doubleValue());
+        costHourDM.setFromFollowRateHour(hourDWD.getOfficialAccountFollowRateHour().doubleValue());
         //TODO:总注册数-----下面全是有问题的
-        costHourDM.setWebRegisterCountTotal(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountTotal());
+        costHourDM.setWebRegisterCountTotal(hourDWD.getOfficialAccountRegisterUserCountTotal());
         //天-总注册数
-        costHourDM.setWebRegisterCountDay(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountDay());
+        costHourDM.setWebRegisterCountDay(hourDWD.getOfficialAccountRegisterUserCountDay());
         //小时-总注册数
-        costHourDM.setWebRegisterCountHour(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountHour());
+        costHourDM.setWebRegisterCountHour(hourDWD.getOfficialAccountRegisterUserCountHour());
         //总注册人数
-        costHourDM.setWebRegisterUvTotal(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountTotal());
+        costHourDM.setWebRegisterUvTotal(hourDWD.getOfficialAccountRegisterUserCountTotal());
         //天-总注册人数
-        costHourDM.setWebRegisterUvDay(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountDay());
+        costHourDM.setWebRegisterUvDay(hourDWD.getOfficialAccountRegisterUserCountDay());
         //小时-总注册人数
-        costHourDM.setWebRegisterUvHour(adStatOfMinuteDWD.getOfficialAccountRegisterUserCountHour());
+        costHourDM.setWebRegisterUvHour(hourDWD.getOfficialAccountRegisterUserCountHour());
         //总平均注册成本
-        costHourDM.setWebRegisterCostTotal(adStatOfMinuteDWD.getOfficialAccountRegisterCostAll());
+        costHourDM.setWebRegisterCostTotal(hourDWD.getOfficialAccountRegisterCostAll());
         //天-总平均注册成本
-        costHourDM.setWebRegisterCostDay(adStatOfMinuteDWD.getOfficialAccountRegisterCostDay());
+        costHourDM.setWebRegisterCostDay(hourDWD.getOfficialAccountRegisterCostDay());
         //小时-总平均注册成本
-        costHourDM.setWebRegisterCostHour(adStatOfMinuteDWD.getOfficialAccountRegisterCostHour());
+        costHourDM.setWebRegisterCostHour(hourDWD.getOfficialAccountRegisterCostHour());
         return costHourDM;
     }
 }