【实战】Flink实现统计每个小时的各类日志条数

【实战】Flink实现统计每个小时的各类日志条数

Flink实现统计每个小时的各类日志条数

SimpleDateFormat
import java.util.{Date, Properties}
import org.apache.flink.api.scala._
le.gson.{JsonObject, JsonParser}
import com.iflytekzunhong.util.RedisPoolUtil
import org.apache.flink.apimon.functions.{AggregateFunction, ReduceFunction}
import org.apache.flink.apimon.serialization.SimpleStringSchema
import org.apache.flink.apimon.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.peutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.iggers.{ContinuousEventTimeTrigger, Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.tors.kafka.FlinkKafkaConsumer010
import org.apache.flink.util.Collector
import org.apache.log4j.{Level, Logger}object FlinkDemand3 {Logger("org").setLevel(Level.ERROR)var logs = ""case cl