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)  //主输出流
          }
      }
    }

  • 相关阅读:
    python 模块和包以及他们的导入关系
    python time,random,os,sys,序列化模块
    python 可迭代对象,迭代器和生成器,lambda表达式
    python之拷贝(深浅)
    python 内置函数
    python 函数的递归
    python 装饰器
    python 函数基础知识整理
    @清晰掉 string.h之基础堵漏
    C和C++之间库的互相调用
  • 原文地址:https://www.cnblogs.com/kpwong/p/14101497.html
Copyright © 2011-2022 走看看