version 0.89

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

@@ -214,7 +214,7 @@ public class AdStatJob {
 
 
         // ------------------------------------------------------- 处理广告的天数据 -----------------------------------------
-        KafkaSource<String> adStreamOfDaySource = KafkaComponent.buildKafkaSource(props, KafkaComponent.KafkaTopic.adDayTopic, KafkaComponent.KafkaTopic.KafkaGroupId.adDayConsumerGroup);
+        /*KafkaSource<String> adStreamOfDaySource = KafkaComponent.buildKafkaSource(props, KafkaComponent.KafkaTopic.adDayTopic, KafkaComponent.KafkaTopic.KafkaGroupId.adDayConsumerGroup);
 
         DataStreamSource<String> adStreamOfDayIn = env.fromSource(adStreamOfDaySource, WatermarkStrategy.noWatermarks(), "adStreamOfMinuteSource_kafka");
 
@@ -353,7 +353,7 @@ public class AdStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfDayDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("adDayDWDYearStream", adDayDWDYearStream.keyBy(AdStatOfDayDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdStatOfDayDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(AdStatOfDayDWD.class));*/
 
         env.execute();
     }

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

@@ -58,9 +58,6 @@ public class PlanStatJob {
     private static final String OLDEST_DAY = "2019-01-01";
 
     public static void main(String[] args) throws Exception {
-        System.setProperty("javax.net.ssl.trustStore", "D:\\Downloads\\kafka.client.truststore.jks");
-        // System.setProperty("javax.net.ssl.trustStore", "/root/flink-1.13.2/kafka.client.truststore.jks");
-        System.setProperty("javax.net.ssl.trustStorePassword", "KafkaOnsClient");
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
         // 加载配置文件到 flink的全局配置中

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

@@ -24,7 +24,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
-import org.apache.flink.table.planner.expressions.In;
 import org.apache.flink.util.Collector;
 
 import java.time.Duration;

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

@@ -118,23 +118,36 @@ public class AdMinuteDWDProcess extends ProcessWindowFunction<AdDataOfMinuteODS,
         System.out.println("输出：" + JsonUtil.toString(newAdStat));
         lastReduceState.put(beginDateTime.format(formatForLastReduceKey), newAdStat);
         yesterdayMinuteState.put(statDay, newAdStat);
-        
+
+        List<String> removeKeys = new ArrayList<>(10);
         for (String key : lastReduceState.keys()) {
             if (key.equals(beginDateTime.format(formatForLastReduceKey))
                     || key.equals(beginDateTime.minusMinutes(5L).format(formatForLastReduceKey))
                     || key.equals(beginDateTime.minusMinutes(10L).format(formatForLastReduceKey))) {
                 continue;
             }
-            lastReduceState.remove(key);
+            removeKeys.add(key);
+        }
+        if (!removeKeys.isEmpty()) {
+            for (String key : removeKeys) {
+                lastReduceState.remove(key);
+            }
         }
+        removeKeys.clear();
         for (String key : yesterdayMinuteState.keys()) {
             if (key.equals(DateUtil.formatLocalDate(beginDate))
                     || key.equals(DateUtil.formatLocalDate(beginDate.minusDays(1L)))
                     || key.equals(DateUtil.formatLocalDate(beginDate.minusDays(2L)))) {
                 continue;
             }
+            removeKeys.add(key);
             yesterdayMinuteState.remove(key);
         }
+        if (!removeKeys.isEmpty()) {
+            for (String key : removeKeys) {
+                yesterdayMinuteState.remove(key);
+            }
+        }
     }
 
     @Override

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

@@ -114,21 +114,33 @@ public class PlanMinuteDWDProcess extends ProcessWindowFunction<AdDataOfMinuteOD
         lastReduceState.put(beginDateTime.format(formatForLastReduceKey), newAdStat);
         yesterdayMinuteState.put(statDay, newAdStat);
 
+        List<String> removeKeys = new ArrayList<>(10);
         for (String key : lastReduceState.keys()) {
             if (key.equals(beginDateTime.format(formatForLastReduceKey))
                     || key.equals(beginDateTime.minusMinutes(5L).format(formatForLastReduceKey))
                     || key.equals(beginDateTime.minusMinutes(10L).format(formatForLastReduceKey))) {
                 continue;
             }
-            lastReduceState.remove(key);
+            removeKeys.add(key);
         }
+        if (!removeKeys.isEmpty()) {
+            for (String key : removeKeys) {
+                lastReduceState.remove(key);
+            }
+        }
+        removeKeys.clear();
         for (String key : yesterdayMinuteState.keys()) {
             if (key.equals(DateUtil.formatLocalDate(beginDate))
                     || key.equals(DateUtil.formatLocalDate(beginDate.minusDays(1L)))
                     || key.equals(DateUtil.formatLocalDate(beginDate.minusDays(2L)))) {
                 continue;
             }
-            yesterdayMinuteState.remove(key);
+            removeKeys.add(key);
+        }
+        if (!removeKeys.isEmpty()) {
+            for (String key : removeKeys) {
+                yesterdayMinuteState.remove(key);
+            }
         }
     }
 

flink-ad-monitoring/src/main/resources/application.properties

@@ -12,4 +12,5 @@ kafka.servers=114.55.59.94:9093,112.124.33.132:9093
 kafka.username=alikafka_pre-cn-tl32fsx4l00x
 kafka.password=VOEdhZLjOrL76lrl5bqPtydtoEkbs0Ny
 kafka.sslPath=D:\\Downloads\\kafka.client.truststore.jks
+# kafka.sslPath=/root/flink-1.13.2/kafka.client.truststore.jks
 kafka.connModule=SASL_SSL