version 0.89 (清理空直)

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

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

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

@@ -21,6 +21,7 @@ import flink.zanxiangnet.ad.monitoring.trigger.AdMinuteODSStreamTrigger;
 import flink.zanxiangnet.ad.monitoring.util.DateUtil;
 import flink.zanxiangnet.ad.monitoring.util.JsonUtil;
 import flink.zanxiangnet.ad.monitoring.kafka.KafkaComponent;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.eventtime.*;
 import org.apache.flink.api.common.state.MapState;
@@ -28,6 +29,7 @@ import org.apache.flink.api.common.state.MapStateDescriptor;
 import org.apache.flink.api.common.state.ValueState;
 import org.apache.flink.api.common.state.ValueStateDescriptor;
 import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -50,6 +52,7 @@ import java.time.LocalTime;
 import java.util.*;
 import java.util.stream.Collectors;
 
+@Slf4j
 public class AdStatJob {
 
     /**
@@ -71,7 +74,7 @@ public class AdStatJob {
         env.getConfig().setGlobalJobParameters(configuration);
 
         KafkaSource<String> adStreamOfMinuteSource = KafkaComponent.buildKafkaSource(props, KafkaComponent.KafkaTopic.adHourTopic, KafkaComponent.KafkaTopic.KafkaGroupId.adHourConsumerGroup);
-        DataStreamSource<String> adStreamOfMinuteIn = env.fromSource(adStreamOfMinuteSource, WatermarkStrategy.noWatermarks(), "adStreamOfMinuteSource_kafka");
+        DataStreamSource<String> adStreamOfMinuteIn = env.fromSource(adStreamOfMinuteSource, WatermarkStrategy.noWatermarks(), "adHourSource_kafka");
 
         // 广告分钟数据（前 5分钟的广告消耗数据）
         final OutputTag<AdDataOfMinuteODS> adMinuteStreamTag = new OutputTag<AdDataOfMinuteODS>("adMinuteStream") {
@@ -101,6 +104,7 @@ public class AdStatJob {
                             adODS.setAccountId(dto.getAccountId());
                             adODS.setAgencyAccountId(dto.getHourlyReportsGetListStruct().getAccountId());
                             adODS.setCreateTime(new Date(createTime));
+                            adODS.removeNull();
                             context.output(adHourStreamTag, adODS);
                         } else {
                             AdDataOfMinuteODS adODS = new AdDataOfMinuteODS();
@@ -111,6 +115,7 @@ public class AdStatJob {
                             adODS.setAccountId(dto.getAccountId());
                             adODS.setAgencyAccountId(dto.getHourlyReportsGetListStruct().getAccountId());
                             adODS.setCreateTime(new Date(createTime));
+                            adODS.removeNull();
                             context.output(adMinuteStreamTag, adODS);
                         }
                     }
@@ -122,7 +127,8 @@ public class AdStatJob {
         // adMinuteODSStream.addSink(new TunnelBatchSink<>(AdDataOfMinuteODS.class, 36000L, 64000L, 3));
         new KeyedBatchStream<>("adMinuteODSStream", adMinuteODSStream.keyBy(AdDataOfMinuteODS::getStatDay), 4000L, 2 * 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdDataOfMinuteODS.class));
+                .addSink(new TunnelBatchStreamSink<>(AdDataOfMinuteODS.class))
+                .name("sink_ad_minute_ods");
 
         SingleOutputStreamOperator<AdStatOfMinuteDWD> adMinuteDWDStream = adMinuteODSStream
                 // 打水印，允许数据延迟 6分钟，同时指定时间流
@@ -136,7 +142,8 @@ public class AdStatJob {
         // .addSink(new TunnelBatchSink<>(AdStatOfMinuteDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("adMinuteDWDStream", adMinuteDWDStream.keyBy(AdStatOfMinuteDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdStatOfMinuteDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(AdStatOfMinuteDWD.class))
+                .name("sink_ad_minute_dwd");
 
         // 小时流（直接写到小时报表的 ods）
         DataStream<AdDataOfHourODS> adHourODSStream = adODSStream.getSideOutput(adHourStreamTag);
@@ -144,7 +151,8 @@ public class AdStatJob {
         // adHourODSStream.addSink(new TunnelBatchSink<>(AdDataOfHourODS.class, 36000L, 64000L, 10));
         new KeyedBatchStream<>("adHourODSStream", adHourODSStream.keyBy(AdDataOfHourODS::getStatDay), 4000L, 5 * 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdDataOfHourODS.class));
+                .addSink(new TunnelBatchStreamSink<>(AdDataOfHourODS.class))
+                .name("sink_ad_hour_ods");
 
         SingleOutputStreamOperator<AdStatOfHourDWD> adHourDWDStream = adHourODSStream.keyBy(AdDataOfHourODS::getAdId)
                 .countWindow(1).process(new ProcessWindowFunction<AdDataOfHourODS, AdStatOfHourDWD, Long, GlobalWindow>() {
@@ -210,13 +218,14 @@ public class AdStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfHourDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("adHourDWDStream", adHourDWDStream.keyBy(AdStatOfHourDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdStatOfHourDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(AdStatOfHourDWD.class))
+                .name("sink_ad_hour_dwd");
 
 
         // ------------------------------------------------------- 处理广告的天数据 -----------------------------------------
-        /*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");
+        DataStreamSource<String> adStreamOfDayIn = env.fromSource(adStreamOfDaySource, WatermarkStrategy.noWatermarks(), "adDaySource_kafka");
 
         // 广告日数据。往前回滚 10天
         final OutputTag<AdDataOfDayODS> adDayStreamRollDayTag = new OutputTag<AdDataOfDayODS>("adDayStreamRollDayTag") {
@@ -239,6 +248,7 @@ public class AdStatJob {
                     adODS.setAdgroupId(struct.getAdgroupId());
                     adODS.setAdId(struct.getAdId());
                     adODS.setCreateTime(createTime);
+                    adODS.removeNull();
                     return AdStatOfDayODSDTO.builder()
                             .startDate(dto.getStartDate())
                             .endDate(dto.getEndDate())
@@ -248,7 +258,8 @@ public class AdStatJob {
         // adDayODSStream.map(AdStatOfDayODSDTO::getAdDataOfDayODS).addSink(new TunnelBatchSink<>(AdDataOfDayODS.class, 36000L, 6000L, 10));
         new KeyedBatchStream<>("adDayODSStream", adDayODSStream.map(AdStatOfDayODSDTO::getAdDataOfDayODS).keyBy(AdDataOfDayODS::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdDataOfDayODS.class));
+                .addSink(new TunnelBatchStreamSink<>(AdDataOfDayODS.class))
+                .name("sink_ad_day_ods");
 
         // 拆分流
         SingleOutputStreamOperator<AdStatOfDayODSDTO> adDayODSStreamSplit = adDayODSStream.process(new ProcessFunction<AdStatOfDayODSDTO, AdStatOfDayODSDTO>() {
@@ -329,7 +340,8 @@ public class AdStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfDayDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("adDayDWDStream", adDayDWDStream.keyBy(AdStatOfDayDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(AdStatOfDayDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(AdStatOfDayDWD.class))
+                .name("sink_ad_day_dwd");
 
         SingleOutputStreamOperator<AdStatOfDayDWD> adDayDWDYearStream = adDayODSStreamSplit.getSideOutput(adDayStreamRollYearTag)
                 .keyBy(AdDataOfDayODS::getAdId)
@@ -353,8 +365,9 @@ 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))
+                .name("sink_ad_year_dwd");
 
-        env.execute();
+        env.execute("job_ad");
     }
 }

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

@@ -71,7 +71,7 @@ public class PlanStatJob {
         env.getConfig().setGlobalJobParameters(configuration);
 
         KafkaSource<String> adStreamOfMinuteSource = KafkaComponent.buildKafkaSource(props, KafkaComponent.KafkaTopic.adHourTopic, KafkaComponent.KafkaTopic.KafkaGroupId.planHourConsumerGroup);
-        DataStreamSource<String> adStreamOfMinuteIn = env.fromSource(adStreamOfMinuteSource, WatermarkStrategy.noWatermarks(), "planStreamOfMinuteSource_kafka");
+        DataStreamSource<String> adStreamOfMinuteIn = env.fromSource(adStreamOfMinuteSource, WatermarkStrategy.noWatermarks(), "planHourSource_kafka");
 
         // 广告分钟数据（前 5分钟的广告消耗数据）
         final OutputTag<AdDataOfMinuteODS> adMinuteStreamTag = new OutputTag<AdDataOfMinuteODS>("adMinuteStream") {
@@ -101,6 +101,7 @@ public class PlanStatJob {
                             adODS.setAccountId(dto.getAccountId());
                             adODS.setAgencyAccountId(dto.getHourlyReportsGetListStruct().getAccountId());
                             adODS.setCreateTime(new Date(createTime));
+                            adODS.removeNull();
                             context.output(adHourStreamTag, adODS);
                         } else {
                             AdDataOfMinuteODS adODS = new AdDataOfMinuteODS();
@@ -111,6 +112,7 @@ public class PlanStatJob {
                             adODS.setAccountId(dto.getAccountId());
                             adODS.setAgencyAccountId(dto.getHourlyReportsGetListStruct().getAccountId());
                             adODS.setCreateTime(new Date(createTime));
+                            adODS.removeNull();
                             context.output(adMinuteStreamTag, adODS);
                         }
                     }
@@ -129,7 +131,8 @@ public class PlanStatJob {
         // .addSink(new TunnelBatchSink<>(AdStatOfMinuteDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("planMinuteDWDStream", planMinuteDWDStream.keyBy(PlanStatOfMinuteDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(PlanStatOfMinuteDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(PlanStatOfMinuteDWD.class))
+                .name("sink_plan_minute_dwd");
 
         // 小时流
         SingleOutputStreamOperator<PlanStatOfHourDWD> planHourDWDStream = adODSStream.getSideOutput(adHourStreamTag).keyBy(AdDataOfHourODS::getCampaignId)
@@ -197,12 +200,13 @@ public class PlanStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfHourDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("planHourDWDStream", planHourDWDStream.keyBy(PlanStatOfHourDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(PlanStatOfHourDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(PlanStatOfHourDWD.class))
+                .name("sink_plan_hour_dwd");
 
         // ------------------------------------------------------- 处理广告的天数据 -----------------------------------------
         KafkaSource<String> adStreamOfDaySource = KafkaComponent.buildKafkaSource(props, KafkaComponent.KafkaTopic.adDayTopic, KafkaComponent.KafkaTopic.KafkaGroupId.planDayConsumerGroup);
 
-        DataStreamSource<String> adStreamOfDayIn = env.fromSource(adStreamOfDaySource, WatermarkStrategy.noWatermarks(), "planStreamOfMinuteSource_kafka");
+        DataStreamSource<String> adStreamOfDayIn = env.fromSource(adStreamOfDaySource, WatermarkStrategy.noWatermarks(), "planDaySource_kafka");
 
         // 广告日数据。往前回滚 10天
         final OutputTag<AdDataOfDayODS> adDayStreamRollDayTag = new OutputTag<AdDataOfDayODS>("adDayStreamRollDayTag") {
@@ -225,6 +229,7 @@ public class PlanStatJob {
                     adODS.setAdgroupId(struct.getAdgroupId());
                     adODS.setAdId(struct.getAdId());
                     adODS.setCreateTime(createTime);
+                    adODS.removeNull();
                     return AdStatOfDayODSDTO.builder()
                             .startDate(dto.getStartDate())
                             .endDate(dto.getEndDate())
@@ -315,7 +320,8 @@ public class PlanStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfDayDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("planDayDWDStream", planDayDWDStream.keyBy(PlanStatOfDayDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(PlanStatOfDayDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(PlanStatOfDayDWD.class))
+                .name("sink_plan_day_dwd");
 
         SingleOutputStreamOperator<PlanStatOfDayDWD> planDayDWDYearStream = adDayODSStreamSplit.getSideOutput(adDayStreamRollYearTag)
                 .keyBy(AdDataOfDayODS::getCampaignId)
@@ -339,8 +345,9 @@ public class PlanStatJob {
         //.addSink(new TunnelBatchSink<>(AdStatOfDayDWD.class, 30000L, 365L, 6));
         new KeyedBatchStream<>("planDayDWDYearStream", planDayDWDYearStream.keyBy(PlanStatOfDayDWD::getStatDay), 4000L, 60 * 1000L)
                 .toBatch()
-                .addSink(new TunnelBatchStreamSink<>(PlanStatOfDayDWD.class));
+                .addSink(new TunnelBatchStreamSink<>(PlanStatOfDayDWD.class))
+                .name("sink_plan_year_dwd");
 
-        env.execute();
+        env.execute("job_plan");
     }
 }

flink-ad-monitoring/src/main/java/flink/zanxiangnet/ad/monitoring/kafka/KafkaComponent.java

@@ -4,6 +4,7 @@ import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
 import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
@@ -27,6 +28,21 @@ public class KafkaComponent {
     public static KafkaSource<String> buildKafkaSource(Properties props, String topic, String groupId) {
         String connModule = props.getProperty(KafkaProperties.KAFKA_MODULE);
         Properties kafkaProps = new Properties();
+        kafkaProps.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
+        //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
+        kafkaProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
+        kafkaProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 30000);
+        kafkaProps.put("max.incremental.fetch.session.cache.slots", 1000);
+
+        //设置单次拉取的量，走公网访问时，该参数会有较大影响。
+        kafkaProps.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
+        kafkaProps.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
+        //每次Poll的最大数量。
+        //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
+        kafkaProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
+
+        //Hostname校验改成空。
+        kafkaProps.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
         if (connModule.equals(KafkaProperties.KAFKA_MODULE_SASL_SSL)) {
             System.setProperty("javax.net.ssl.trustStore", props.getProperty(KafkaProperties.KAFKA_SSL_PATH));
             System.setProperty("javax.net.ssl.trustStorePassword", "KafkaOnsClient");

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

@@ -505,4 +505,298 @@ public class AdDataOfDayODS implements Serializable {
      */
     @SerializedName("no_interest_count")
     private Long noInterestCount;
+
+    public void removeNull() {
+        if (costDeviationRate == null) {
+            costDeviationRate = 0.0;
+        }
+
+        if (cost == null) {
+            cost = 0L;
+        }
+
+        if (compensationAmount == null) {
+            compensationAmount = 0L;
+        }
+
+        if (viewCount == null) {
+            viewCount = 0L;
+        }
+
+        if (thousandDisplayPrice == null) {
+            thousandDisplayPrice = 0L;
+        }
+
+        if (viewUserCount == null) {
+            viewUserCount = 0L;
+        }
+
+        if (avgViewPerUser == null) {
+            avgViewPerUser = 0.0;
+        }
+
+        if (validClickCount == null) {
+            validClickCount = 0L;
+        }
+
+        if (clickUserCount == null) {
+            clickUserCount = 0L;
+        }
+
+        if (ctr == null) {
+            ctr = 0.0;
+        }
+
+        if (cpc == null) {
+            cpc = 0L;
+        }
+
+        if (valuableClickCount == null) {
+            valuableClickCount = 0L;
+        }
+
+        if (valuableClickRate == null) {
+            valuableClickRate = 0.0;
+        }
+
+        if (valuableClickCost == null) {
+            valuableClickCost = 0L;
+        }
+
+        if (conversionsCount == null) {
+            conversionsCount = 0L;
+        }
+
+        if (conversionsCost == null) {
+            conversionsCost = 0L;
+        }
+
+        if (conversionsRate == null) {
+            conversionsRate = 0.0;
+        }
+
+        if (deepConversionsCount == null) {
+            deepConversionsCount = 0L;
+        }
+
+        if (deepConversionsCost == null) {
+            deepConversionsCost = 0L;
+        }
+
+        if (deepConversionsRate == null) {
+            deepConversionsRate = 0.0;
+        }
+
+        if (keyPageUv == null) {
+            keyPageUv = 0L;
+        }
+
+        if (orderCount == null) {
+            orderCount = 0L;
+        }
+
+        if (firstDayOrderCount == null) {
+            firstDayOrderCount = 0L;
+        }
+
+        if (webOrderCost == null) {
+            webOrderCost = 0L;
+        }
+
+        if (orderRate == null) {
+            orderRate = 0.0;
+        }
+
+        if (orderAmount == null) {
+            orderAmount = 0L;
+        }
+
+        if (firstDayOrderAmount == null) {
+            firstDayOrderAmount = 0L;
+        }
+
+        if (orderUnitPrice == null) {
+            orderUnitPrice = 0L;
+        }
+
+        if (orderRoi == null) {
+            orderRoi = 0.0;
+        }
+
+        if (signInCount == null) {
+            signInCount = 0L;
+        }
+
+        if (addWishlistCount == null) {
+            addWishlistCount = 0L;
+        }
+
+        if (viewCommodityPageUv == null) {
+            viewCommodityPageUv = 0L;
+        }
+
+        if (pageReservationCount == null) {
+            pageReservationCount = 0L;
+        }
+
+        if (leadsPurchaseUv == null) {
+            leadsPurchaseUv = 0L;
+        }
+
+        if (leadsPurchaseCost == null) {
+            leadsPurchaseCost = 0L;
+        }
+
+        if (leadsPurchaseRate == null) {
+            leadsPurchaseRate = 0.0;
+        }
+
+        if (scanFollowCount == null) {
+            scanFollowCount = 0L;
+        }
+
+        if (wechatAppRegisterUv == null) {
+            wechatAppRegisterUv = 0L;
+        }
+
+        if (wechatMinigameRegisterCost == null) {
+            wechatMinigameRegisterCost = 0L;
+        }
+
+        if (wechatMinigameRegisterRate == null) {
+            wechatMinigameRegisterRate = 0.0;
+        }
+
+        if (wechatMinigameArpu == null) {
+            wechatMinigameArpu = 0.0;
+        }
+
+        if (wechatMinigameRetentionCount == null) {
+            wechatMinigameRetentionCount = 0L;
+        }
+
+        if (wechatMinigameCheckoutCount == null) {
+            wechatMinigameCheckoutCount = 0L;
+        }
+
+        if (wechatMinigameCheckoutAmount == null) {
+            wechatMinigameCheckoutAmount = 0L;
+        }
+
+        if (officialAccountFollowCount == null) {
+            officialAccountFollowCount = 0L;
+        }
+
+        if (officialAccountFollowCost == null) {
+            officialAccountFollowCost = 0L;
+        }
+
+        if (officialAccountFollowRate == null) {
+            officialAccountFollowRate = 0.0;
+        }
+
+        if (officialAccountRegisterUserCount == null) {
+            officialAccountRegisterUserCount = 0L;
+        }
+
+        if (officialAccountRegisterRate == null) {
+            officialAccountRegisterRate = 0.0;
+        }
+
+        if (officialAccountRegisterCost == null) {
+            officialAccountRegisterCost = 0L;
+        }
+
+        if (officialAccountRegisterAmount == null) {
+            officialAccountRegisterAmount = 0L;
+        }
+
+        if (officialAccountRegisterRoi == null) {
+            officialAccountRegisterRoi = 0L;
+        }
+
+        if (officialAccountApplyCount == null) {
+            officialAccountApplyCount = 0L;
+        }
+
+        if (officialAccountApplyUserCount == null) {
+            officialAccountApplyUserCount = 0L;
+        }
+
+        if (officialAccountApplyRate == null) {
+            officialAccountApplyRate = 0.0;
+        }
+
+        if (officialAccountApplyCost == null) {
+            officialAccountApplyCost = 0L;
+        }
+
+        if (officialAccountApplyAmount == null) {
+            officialAccountApplyAmount = 0L;
+        }
+
+        if (officialAccountApplyRoi == null) {
+            officialAccountApplyRoi = 0L;
+        }
+
+        if (officialAccountOrderCount == null) {
+            officialAccountOrderCount = 0L;
+        }
+
+        if (officialAccountFirstDayOrderCount == null) {
+            officialAccountFirstDayOrderCount = 0L;
+        }
+
+        if (officialAccountOrderUserCount == null) {
+            officialAccountOrderUserCount = 0L;
+        }
+
+        if (officialAccountOrderRate == null) {
+            officialAccountOrderRate = 0.0;
+        }
+
+        if (officialAccountOrderCost == null) {
+            officialAccountOrderCost = 0L;
+        }
+
+        if (officialAccountOrderAmount == null) {
+            officialAccountOrderAmount = 0L;
+        }
+
+        if (officialAccountFirstDayOrderAmount == null) {
+            officialAccountFirstDayOrderAmount = 0L;
+        }
+
+        if (officialAccountOrderRoi == null) {
+            officialAccountOrderRoi = 0L;
+        }
+
+        if (officialAccountConsultCount == null) {
+            officialAccountConsultCount = 0L;
+        }
+
+        if (officialAccountReaderCount == null) {
+            officialAccountReaderCount = 0L;
+        }
+
+        if (officialAccountCreditApplyUserCount == null) {
+            officialAccountCreditApplyUserCount = 0L;
+        }
+
+        if (officialAccountCreditUserCount == null) {
+            officialAccountCreditUserCount = 0L;
+        }
+
+        if (forwardCount == null) {
+            forwardCount = 0L;
+        }
+
+        if (forwardUserCount == null) {
+            forwardUserCount = 0L;
+        }
+
+        if (noInterestCount == null) {
+            noInterestCount = 0L;
+        }
+    }
 }

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

@@ -467,4 +467,202 @@ public class AdDataOfHourODS implements Serializable {
      */
     @SerializedName("no_interest_count")
     private Long noInterestCount;
+
+    /**
+     * 清理空值
+     */
+    public void removeNull() {
+        if (costDeviationRate == null) {
+            costDeviationRate = 0.0;
+        }
+        if (cost == null) {
+            cost = 0L;
+        }
+        if (compensationAmount == null) {
+            compensationAmount = 0L;
+        }
+        if (viewCount == null) {
+            viewCount = 0L;
+        }
+        if (thousandDisplayPrice == null) {
+            thousandDisplayPrice = 0L;
+        }
+        if (avgViewPerUser == null) {
+            avgViewPerUser = 0.0;
+        }
+        if (validClickCount == null) {
+            validClickCount = 0L;
+        }
+        if (ctr == null) {
+            ctr = 0.0;
+        }
+        if (cpc == null) {
+            cpc = 0L;
+        }
+        if (valuableClickCount == null) {
+            valuableClickCount = 0L;
+        }
+        if (valuableClickRate == null) {
+            valuableClickRate = 0.0;
+        }
+        if (valuableClickCost == null) {
+            valuableClickCost = 0L;
+        }
+        if (conversionsCount == null) {
+            conversionsCount = 0L;
+        }
+        if (conversionsCost == null) {
+            conversionsCost = 0L;
+        }
+        if (conversionsRate == null) {
+            conversionsRate = 0.0;
+        }
+        if (deepConversionsCount == null) {
+            deepConversionsCount = 0L;
+        }
+        if (deepConversionsCost == null) {
+            deepConversionsCost = 0L;
+        }
+        if (deepConversionsRate == null) {
+            deepConversionsRate = 0.0;
+        }
+        if (orderCount == null) {
+            orderCount = 0L;
+        }
+        if (firstDayOrderCount == null) {
+            firstDayOrderCount = 0L;
+        }
+        if (webOrderCost == null) {
+            webOrderCost = 0L;
+        }
+        if (orderRate == null) {
+            orderRate = 0.0;
+        }
+        if (orderAmount == null) {
+            orderAmount = 0L;
+        }
+        if (firstDayOrderAmount == null) {
+            firstDayOrderAmount = 0L;
+        }
+        if (orderUnitPrice == null) {
+            orderUnitPrice = 0L;
+        }
+        if (orderRoi == null) {
+            orderRoi = 0.0;
+        }
+        if (signInCount == null) {
+            signInCount = 0L;
+        }
+        if (scanFollowCount == null) {
+            scanFollowCount = 0L;
+        }
+        if (wechatAppRegisterUv == null) {
+            wechatAppRegisterUv = 0L;
+        }
+        if (wechatMinigameRegisterCost == null) {
+            wechatMinigameRegisterCost = 0L;
+        }
+        if (wechatMinigameRegisterRate == null) {
+            wechatMinigameRegisterRate = 0.0;
+        }
+        if (wechatMinigameArpu == null) {
+            wechatMinigameArpu = 0.0;
+        }
+        if (wechatMinigameRetentionCount == null) {
+            wechatMinigameRetentionCount = 0L;
+        }
+        if (wechatMinigameCheckoutCount == null) {
+            wechatMinigameCheckoutCount = 0L;
+        }
+        if (wechatMinigameCheckoutAmount == null) {
+            wechatMinigameCheckoutAmount = 0L;
+        }
+        if (officialAccountFollowCount == null) {
+            officialAccountFollowCount = 0L;
+        }
+        if (officialAccountFollowCost == null) {
+            officialAccountFollowCost = 0L;
+        }
+        if (officialAccountFollowRate == null) {
+            officialAccountFollowRate = 0.0;
+        }
+        if (officialAccountRegisterUserCount == null) {
+            officialAccountRegisterUserCount = 0L;
+        }
+        if (officialAccountRegisterRate == null) {
+            officialAccountRegisterRate = 0.0;
+        }
+        if (officialAccountRegisterCost == null) {
+            officialAccountRegisterCost = 0L;
+        }
+        if (officialAccountRegisterAmount == null) {
+            officialAccountRegisterAmount = 0L;
+        }
+        if (officialAccountRegisterRoi == null) {
+            officialAccountRegisterRoi = 0L;
+        }
+        if (officialAccountApplyCount == null) {
+            officialAccountApplyCount = 0L;
+        }
+        if (officialAccountApplyUserCount == null) {
+            officialAccountApplyUserCount = 0L;
+        }
+        if (officialAccountApplyRate == null) {
+            officialAccountApplyRate = 0.0;
+        }
+        if (officialAccountApplyCost == null) {
+            officialAccountApplyCost = 0L;
+        }
+        if (officialAccountApplyAmount == null) {
+            officialAccountApplyAmount = 0L;
+        }
+        if (officialAccountApplyRoi == null) {
+            officialAccountApplyRoi = 0L;
+        }
+        if (officialAccountOrderCount == null) {
+            officialAccountOrderCount = 0L;
+        }
+        if (officialAccountFirstDayOrderCount == null) {
+            officialAccountFirstDayOrderCount = 0L;
+        }
+        if (officialAccountOrderUserCount == null) {
+            officialAccountOrderUserCount = 0L;
+        }
+        if (officialAccountOrderRate == null) {
+            officialAccountOrderRate = 0.0;
+        }
+        if (officialAccountOrderCost == null) {
+            officialAccountOrderCost = 0L;
+        }
+        if (officialAccountOrderAmount == null) {
+            officialAccountOrderAmount = 0L;
+        }
+        if (officialAccountFirstDayOrderAmount == null) {
+            officialAccountFirstDayOrderAmount = 0L;
+        }
+        if (officialAccountOrderRoi == null) {
+            officialAccountOrderRoi = 0L;
+        }
+        if (officialAccountConsultCount == null) {
+            officialAccountConsultCount = 0L;
+        }
+        if (officialAccountReaderCount == null) {
+            officialAccountReaderCount = 0L;
+        }
+        if (officialAccountCreditApplyUserCount == null) {
+            officialAccountCreditApplyUserCount = 0L;
+        }
+        if (officialAccountCreditUserCount == null) {
+            officialAccountCreditUserCount = 0L;
+        }
+        if (forwardCount == null) {
+            forwardCount = 0L;
+        }
+        if (forwardUserCount == null) {
+            forwardUserCount = 0L;
+        }
+        if (noInterestCount == null) {
+            noInterestCount = 0L;
+        }
+    }
 }

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

@@ -468,4 +468,202 @@ public class AdDataOfMinuteODS implements Serializable {
      */
     @SerializedName("no_interest_count")
     private Long noInterestCount;
+
+    /**
+     * 清理空值
+     */
+    public void removeNull() {
+        if (costDeviationRate == null) {
+            costDeviationRate = 0.0;
+        }
+        if (cost == null) {
+            cost = 0L;
+        }
+        if (compensationAmount == null) {
+            compensationAmount = 0L;
+        }
+        if (viewCount == null) {
+            viewCount = 0L;
+        }
+        if (thousandDisplayPrice == null) {
+            thousandDisplayPrice = 0L;
+        }
+        if (avgViewPerUser == null) {
+            avgViewPerUser = 0.0;
+        }
+        if (validClickCount == null) {
+            validClickCount = 0L;
+        }
+        if (ctr == null) {
+            ctr = 0.0;
+        }
+        if (cpc == null) {
+            cpc = 0L;
+        }
+        if (valuableClickCount == null) {
+            valuableClickCount = 0L;
+        }
+        if (valuableClickRate == null) {
+            valuableClickRate = 0.0;
+        }
+        if (valuableClickCost == null) {
+            valuableClickCost = 0L;
+        }
+        if (conversionsCount == null) {
+            conversionsCount = 0L;
+        }
+        if (conversionsCost == null) {
+            conversionsCost = 0L;
+        }
+        if (conversionsRate == null) {
+            conversionsRate = 0.0;
+        }
+        if (deepConversionsCount == null) {
+            deepConversionsCount = 0L;
+        }
+        if (deepConversionsCost == null) {
+            deepConversionsCost = 0L;
+        }
+        if (deepConversionsRate == null) {
+            deepConversionsRate = 0.0;
+        }
+        if (orderCount == null) {
+            orderCount = 0L;
+        }
+        if (firstDayOrderCount == null) {
+            firstDayOrderCount = 0L;
+        }
+        if (webOrderCost == null) {
+            webOrderCost = 0L;
+        }
+        if (orderRate == null) {
+            orderRate = 0.0;
+        }
+        if (orderAmount == null) {
+            orderAmount = 0L;
+        }
+        if (firstDayOrderAmount == null) {
+            firstDayOrderAmount = 0L;
+        }
+        if (orderUnitPrice == null) {
+            orderUnitPrice = 0L;
+        }
+        if (orderRoi == null) {
+            orderRoi = 0.0;
+        }
+        if (signInCount == null) {
+            signInCount = 0L;
+        }
+        if (scanFollowCount == null) {
+            scanFollowCount = 0L;
+        }
+        if (wechatAppRegisterUv == null) {
+            wechatAppRegisterUv = 0L;
+        }
+        if (wechatMinigameRegisterCost == null) {
+            wechatMinigameRegisterCost = 0L;
+        }
+        if (wechatMinigameRegisterRate == null) {
+            wechatMinigameRegisterRate = 0.0;
+        }
+        if (wechatMinigameArpu == null) {
+            wechatMinigameArpu = 0.0;
+        }
+        if (wechatMinigameRetentionCount == null) {
+            wechatMinigameRetentionCount = 0L;
+        }
+        if (wechatMinigameCheckoutCount == null) {
+            wechatMinigameCheckoutCount = 0L;
+        }
+        if (wechatMinigameCheckoutAmount == null) {
+            wechatMinigameCheckoutAmount = 0L;
+        }
+        if (officialAccountFollowCount == null) {
+            officialAccountFollowCount = 0L;
+        }
+        if (officialAccountFollowCost == null) {
+            officialAccountFollowCost = 0L;
+        }
+        if (officialAccountFollowRate == null) {
+            officialAccountFollowRate = 0.0;
+        }
+        if (officialAccountRegisterUserCount == null) {
+            officialAccountRegisterUserCount = 0L;
+        }
+        if (officialAccountRegisterRate == null) {
+            officialAccountRegisterRate = 0.0;
+        }
+        if (officialAccountRegisterCost == null) {
+            officialAccountRegisterCost = 0L;
+        }
+        if (officialAccountRegisterAmount == null) {
+            officialAccountRegisterAmount = 0L;
+        }
+        if (officialAccountRegisterRoi == null) {
+            officialAccountRegisterRoi = 0L;
+        }
+        if (officialAccountApplyCount == null) {
+            officialAccountApplyCount = 0L;
+        }
+        if (officialAccountApplyUserCount == null) {
+            officialAccountApplyUserCount = 0L;
+        }
+        if (officialAccountApplyRate == null) {
+            officialAccountApplyRate = 0.0;
+        }
+        if (officialAccountApplyCost == null) {
+            officialAccountApplyCost = 0L;
+        }
+        if (officialAccountApplyAmount == null) {
+            officialAccountApplyAmount = 0L;
+        }
+        if (officialAccountApplyRoi == null) {
+            officialAccountApplyRoi = 0L;
+        }
+        if (officialAccountOrderCount == null) {
+            officialAccountOrderCount = 0L;
+        }
+        if (officialAccountFirstDayOrderCount == null) {
+            officialAccountFirstDayOrderCount = 0L;
+        }
+        if (officialAccountOrderUserCount == null) {
+            officialAccountOrderUserCount = 0L;
+        }
+        if (officialAccountOrderRate == null) {
+            officialAccountOrderRate = 0.0;
+        }
+        if (officialAccountOrderCost == null) {
+            officialAccountOrderCost = 0L;
+        }
+        if (officialAccountOrderAmount == null) {
+            officialAccountOrderAmount = 0L;
+        }
+        if (officialAccountFirstDayOrderAmount == null) {
+            officialAccountFirstDayOrderAmount = 0L;
+        }
+        if (officialAccountOrderRoi == null) {
+            officialAccountOrderRoi = 0L;
+        }
+        if (officialAccountConsultCount == null) {
+            officialAccountConsultCount = 0L;
+        }
+        if (officialAccountReaderCount == null) {
+            officialAccountReaderCount = 0L;
+        }
+        if (officialAccountCreditApplyUserCount == null) {
+            officialAccountCreditApplyUserCount = 0L;
+        }
+        if (officialAccountCreditUserCount == null) {
+            officialAccountCreditUserCount = 0L;
+        }
+        if (forwardCount == null) {
+            forwardCount = 0L;
+        }
+        if (forwardUserCount == null) {
+            forwardUserCount = 0L;
+        }
+        if (noInterestCount == null) {
+            noInterestCount = 0L;
+        }
+    }
 }

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

@@ -1279,7 +1279,7 @@ public class AdStatOfHourDWD {
 
     public static AdStatOfHourDWD reduce(AdStatOfHourDWD value1, AdDataOfHourODS value2, long createTime) {
         AdStatOfHourDWD result = new AdStatOfHourDWD();
-        BeanUtils.copyProperties(result, value2);
+        BeanUtils.copyProperties(value2, result);
         result.setCreateTime(new Date(createTime));
         if (value1 == null) {
             result.setCompensationAmountTotal(value2.getCompensationAmount());

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

@@ -1217,7 +1217,7 @@ public class PlanStatOfHourDWD {
 
     public static PlanStatOfHourDWD reduce(PlanStatOfHourDWD value1, AdDataOfHourODS value2, long createTime) {
         PlanStatOfHourDWD result = new PlanStatOfHourDWD();
-        BeanUtils.copyProperties(result, value2);
+        BeanUtils.copyProperties(value2, result);
         result.setCreateTime(new Date(createTime));
         if (value1 == null) {
             Map<Long, Set<Long>> adGroupMap = new HashMap<>();

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

@@ -11,6 +11,6 @@ maxCompute.tunnelEndpoint=http://dt.cn-hangzhou.maxcompute.aliyun.com
 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.sslPath=D:\\Downloads\\kafka.client.truststore.jks
+kafka.sslPath=/root/flink-1.13.2/kafka.client.truststore.jks
 kafka.connModule=SASL_SSL