Flink学习(十七) Emitting to Side Outputs(侧输出)

我们在生产实践中经常会遇到这样的场景,需把输入源按照需要进行拆分,比如我期望把订单流按照金额大小进行拆分,或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢?

大部分的DataStream API的算子的输出时单一输出,也就是某种数据类型的流。除了split算子(使用split切分过的流是不能被二次切分的),可以将一条流分成多条流,这些流的数据类型也都相同。processfunction的side outputs功能可以产生多条流,并且这些流的数据类型可以不一样。一个side output可以定义为OutputTag[X]对象,X是输出流的数据类型。processfunction可以通过Context对象发送一个事件到一个或者多个sideouputs.

SideOutPut 分流
SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法,在使用 SideOutPut 时,需要按照以下步骤进行:

定义 OutputTag
调用特定函数进行数据拆分
ProcessFunction
KeyedProcessFunction
CoProcessFunction
KeyedCoProcessFunction
ProcessWindowFunction
ProcessAllWindowFunction
在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut:


package com.wyh.processFunctionApi

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
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.getExecutionEnvironment

env.setParallelism(1)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val stream = env.socketTextStream("localhost", 7777)

//Transform操作
val dataStream: DataStream[SensorReading] = stream.map(data => {
val dataArray = data.split(",")
SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
})
//===到来的数据是升序的,准时发车,用assignAscendingTimestamps
//指定哪个字段是时间戳 需要的是毫秒 * 1000
// .assignAscendingTimestamps(_.timestamp * 1000)
//===处理乱序数据
// .assignTimestampsAndWatermarks(new MyAssignerPeriodic())
//==底层也是周期性生成的一个方法 处理乱序数据 延迟1秒种生成水位 同时分配水位和时间戳 括号里传的是等待延迟的时间
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
override def extractTimestamp(t: SensorReading): Long = {
t.timestamp * 1000
}
})

val processedStream = dataStream.process(new FreezingAlert())

//这里打印的是主流
processedStream.print("process data")
//打印侧输出流
processedStream.getSideOutput(new OutputTag[String]("Freezing alert")).print()
processedStream.getSideOutput(new OutputTag[String]("commen data")).print()


//dataStream.print("input data")


env.execute("window Test")

}
}

/**
* 冰点报警 如果小于32F,输出报警信息到侧输出流
*/
//输出的类型是主输出流的数据类型
class FreezingAlert() extends ProcessFunction[SensorReading, SensorReading] {
lazy val alertOutput: OutputTag[String] = new OutputTag[String]("Freezing alert")
lazy val commenOutput: OutputTag[String] = new OutputTag[String]("commen data")

override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
if (value.temperature < 32.0) {
//侧输出流
ctx.output(alertOutput, value.id + "低温报警!!!此时温度为:" + value.temperature)
} else if (value.temperature >= 32.0) {
ctx.output(commenOutput, value.id + "正常温度。。此时温度为:" + value.temperature)
} else {
//主流
out.collect(value)
}
}
}
 

在Linux命令行中输入 nc -lk 7777开启一个服务

Flink学习(十七) Emitting to Side Outputs(侧输出)

 

 输入数据:

Flink学习(十七) Emitting to Side Outputs(侧输出)

 

注意:在主程序中,直接print()打印的主输出流,想要打印侧输出流:

    //这里打印的是主流
    processedStream.print("process data")
    //打印侧输出流
    processedStream.getSideOutput(new OutputTag[String]("Freezing alert")).print()
    processedStream.getSideOutput(new OutputTag[String]("commen data")).print()

 

上一篇:大数据之输出报警信息


下一篇:深度学习各优化算法