标签：输出 flink val DEMO Flink SensorReading org apache import

代码:

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

标签：输出,flink,val,DEMO,Flink,SensorReading,org,apache,import
来源： https://www.cnblogs.com/kpwong/p/14101497.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。