version 0.89 (修改部分 bug）

flink-ad-monitoring/dependency-reduced-pom.xml

@@ -88,7 +88,7 @@
               </filters>
               <transformers>
                 <transformer>
-                  <mainClass>flink.zanxiangnet.ad.monitoring.AdStatJob</mainClass>
+                  <mainClass>flink.zanxiangnet.ad.monitoring.PlanStatJob</mainClass>
                 </transformer>
               </transformers>
             </configuration>

flink-ad-monitoring/pom.xml

@@ -202,7 +202,7 @@ under the License.
                             <transformers>
                                 <transformer
                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                                    <mainClass>flink.zanxiangnet.ad.monitoring.AdStatJob</mainClass>
+                                    <mainClass>flink.zanxiangnet.ad.monitoring.PlanStatJob</mainClass>
                                 </transformer>
                             </transformers>
                         </configuration>

flink-ad-monitoring/src/main/java/flink/zanxiangnet/ad/monitoring/PlanStatJob.java

@@ -259,7 +259,7 @@ public class PlanStatJob {
                     private MapState<String, PlanStatOfDayDWD> historyReduceState;
 
                     @Override
-                    public void open(Configuration conf) {
+                    public void open(Configuration conf) throws Exception {
                         Map<String, String> params = getRuntimeContext()
                                 .getExecutionConfig()
                                 .getGlobalJobParameters()
@@ -278,6 +278,9 @@ public class PlanStatJob {
                     @Override
                     public void process(Long elementsCount, ProcessWindowFunction<AdDataOfDayODS, PlanStatOfDayDWD, Long, GlobalWindow>.Context context,
                                         Iterable<AdDataOfDayODS> iterable, Collector<PlanStatOfDayDWD> collector) throws Exception {
+                        for (String key : historyReduceState.keys()) {
+                            PlanStatOfDayDWD newStatData = historyReduceState.get(key);
+                        }
                         AdDataOfDayODS element = iterable.iterator().next();
                         LocalDate statDay = DateUtil.parseLocalDate(element.getStatDay());
                         long now = System.currentTimeMillis();

flink-ad-monitoring/src/main/java/flink/zanxiangnet/ad/monitoring/pojo/entity/PlanStatOfDayDWD.java

@@ -1134,10 +1134,12 @@ public class PlanStatOfDayDWD implements Serializable {
             result.setNoInterestCountTotal(value2.getNoInterestCount());
             result.setNoInterestCountDay(value2.getNoInterestCount());
         } else {
-            value1.getAdGroupMap().computeIfAbsent(value2.getAdgroupId(), key -> new HashSet<>(3)).add(value2.getAdId());
-            value1.setAdGroupMapStr(PlanUtil.adGroupMapStr(value1.getAdGroupMap()));
-            value1.getAdIds().add(value2.getAdId());
-            value1.setAdIdsStr(PlanUtil.adIdsStr(value1.getAdIds()));
+            result.setAdGroupMap(value1.getAdGroupMap());
+            result.setAdIds(value1.getAdIds());
+            result.getAdGroupMap().computeIfAbsent(value2.getAdgroupId(), key -> new HashSet<>(3)).add(value2.getAdId());
+            result.setAdGroupMapStr(PlanUtil.adGroupMapStr(value1.getAdGroupMap()));
+            result.getAdIds().add(value2.getAdId());
+            result.setAdIdsStr(PlanUtil.adIdsStr(value1.getAdIds()));
 
             result.setCostDeviationRateTotal(value1.getCostDeviationRateTotal() + value2.getCostDeviationRate());
             result.setCostDeviationRateDay(value2.getCostDeviationRate());