zoukankan      html  css  js  c++  java
  • Flink 侧输出流 DEMO

    代码:

    package com.kpwong.sideoutput
    
    import com.atguigu.window.{SensorReading}
    import org.apache.flink.streaming.api.functions.ProcessFunction
    //import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.scala._
    //import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.util.Collector
    
    object SideOutPutTest {
      def main(args: Array[String]): Unit = {
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    
        val SocketDS: DataStream[String] = env.socketTextStream("hadoop202",9999)
    
        val sensorStream: DataStream[SensorReading] = SocketDS.map(data => {
          val dataArray: Array[String] = data.split(",")
          SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
        }).process(new FreeAlert())
    //    val processDS: DataStream[SensorReading] = sensorStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
    //      override def extractTimestamp(element: SensorReading) = {
    //        element.timestamp * 1000
    //      }
    //    }).process(new FreeAlert())
    
        sensorStream.print("main process flow")
    
        sensorStream.getSideOutput(new OutputTag[SensorReading]("zeroAlert")).print("Side out put flow")
    
    
        env.execute()
      }
    }
    
    //冰点报警,如果小于0度,就输出到侧数据流
    class FreeAlert extends  ProcessFunction[SensorReading,SensorReading]{
      lazy  val alertOutPut:OutputTag[SensorReading] = new OutputTag[SensorReading]("zeroAlert")
      override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
        if(value.temperature  < 0.0)
          {
            //小于零度,到侧输出流
            ctx.output(alertOutPut,value)
          }
        else
          {
            out.collect(value)  //主输出流
          }
      }
    }

  • 相关阅读:
    【转】Windows Socket网络编程(二)套接字编程原理
    获取本地IP地址,并在IP CONTROL控件中显示出来
    PAT 1021 Deepest Root[并查集、dfs][难]
    1025 PAT Ranking[排序][一般]
    PAT 1028 List Sorting[排序][一般]
    PAT 1023 Have Fun with Numbers[大数乘法][一般]
    PAT 1026 Table Tennis[比较难]
    Andrew NgML第十章应用机器学习的建议
    PAT 1020 Tree Traversals[二叉树遍历]
    PAT 1022 Digital Library[map使用]
  • 原文地址:https://www.cnblogs.com/kpwong/p/14101497.html
Copyright © 2011-2022 走看看