Shulou(Shulou.com)06/03 Report--

The data with a sliding window function size of 30 seconds, an interval of 15 seconds, and more than 10 seconds of the window is discarded. (the three values of the actual business should be 10 minutes, 1 minute, 5 minutes). Record the code first.

Public static void main (String [] arg) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); env.getConfig (). EnableSysoutLogging (); / / Open Sysout log env.setStreamTimeCharacteristic (TimeCharacteristic.EventTime); / / set the time unit of the window to process time Properties props = new Properties (); props.put ("bootstrap.servers", "kafkaip:9092"); props.put ("group.id", "metric-group4") Props.put ("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); / / key deserializes props.put ("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put ("auto.offset.reset", "earliest") / / value deserialization DataStreamSource dataStreamSource = env.addSource (new FlinkKafkaConsumer011 ("im-message-topic3", / / kafka topic new SimpleStringSchema (), / / String serialization props)) .setParallelism (1) DataStream bean3DataStream = dataStreamSource.map (new MapFunction () {@ Override public Message map (String value) throws Exception {logger.info ("receive msg:" + value); JSONObject jsonObject = JSONObject.parseObject (value) Message s = new Message (jsonObject.getString ("sessionId"), jsonObject.getString ("fromUid"), jsonObject.getString ("toUid"), jsonObject.getString ("chatType"), jsonObject.getString ("type") JsonObject.getString ("msgId"), jsonObject.getString ("msg"), jsonObject.getLong ("timestampSend")) Return s;}}) / / set watermark And filter the data DataStream bean3DataStreamWithAssignTime = bean3DataStream.assignTimestampsAndWatermarks (new TruckTimestamp ()) .timeWindowAll (Time.seconds (30), Time.seconds (15)) .apply (new AllWindowFunction ()) {@ Override public void apply (TimeWindow window, Iterable values) Collector out) throws Exception {for (Message t: values) {logger.info ("window start time:" + new Date (window.getStart ()) .toString ()) Logger.info ("real time:" + new Date (t.getTimestampSend ()). ToString ()); if (t.getTimestampSend ())

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.