??除了來自數(shù)據(jù)流算子的主流結(jié)果輸出之外,可以產(chǎn)生任意數(shù)量的流旁路輸出結(jié)果。旁路輸出結(jié)果數(shù)據(jù)類型與主流結(jié)果的數(shù)據(jù)類型以及其他旁路輸出結(jié)果數(shù)據(jù)類型可以是完全不同的。當(dāng)你需要分割數(shù)據(jù)流時(shí),這個(gè)算子非常有用。通常需要復(fù)制流,然后從每個(gè)數(shù)據(jù)流中過濾掉不需要的數(shù)據(jù)。
?當(dāng)使用旁路輸出時(shí),首先需要定義一個(gè)OutputTag來標(biāo)識(shí)一個(gè)旁路輸出流。
Java
西疇ssl適用于網(wǎng)站、小程序/APP、API接口等需要進(jìn)行數(shù)據(jù)傳輸應(yīng)用場(chǎng)景,ssl證書未來市場(chǎng)廣闊!成為創(chuàng)新互聯(lián)的ssl證書銷售渠道,可以享受市場(chǎng)價(jià)格4-6折優(yōu)惠!如果有意向歡迎電話聯(lián)系或者加微信:13518219792(備注:SSL證書合作)期待與您的合作!
// this needs to be an anonymous inner class, so that we can analyze the type
OutputTag outputTag = new OutputTag("side-output") {};
Scala
val outputTag = OutputTag[String]("side-output")
??注意OutputTag是如何根據(jù)旁路輸出流包含的元素類型typed的。
?可以通過以下函數(shù)發(fā)射數(shù)據(jù)到旁路輸出。
??可以使用Context參數(shù)(在上述函數(shù)中向用戶暴露)將數(shù)據(jù)發(fā)送到OutputTag標(biāo)識(shí)的旁路輸出。以下是從ProcessFunction發(fā)出旁路輸出數(shù)據(jù)的示例:
Java:
DataStream input = ...;
final OutputTag outputTag = new OutputTag("side-output"){};
SingleOutputStreamOperator mainDataStream = input
.process(new ProcessFunction() {
@Override
public void processElement(
Integer value,
Context ctx,
Collector out) throws Exception {
// emit data to regular output
out.collect(value);
// emit data to side output
ctx.output(outputTag, "sideout-" + String.valueOf(value));
}
});
Scala:
val input: DataStream[Int] = ...
val outputTag = OutputTag[String]("side-output")
val mainDataStream = input
.process(new ProcessFunction[Int, Int] {
override def processElement(
value: Int,
ctx: ProcessFunction[Int, Int]#Context,
out: Collector[Int]): Unit = {
// emit data to regular output
out.collect(value)
// emit data to side output
ctx.output(outputTag, "sideout-" + String.valueOf(value))
}
})
要讀取旁路輸出流,在數(shù)據(jù)流運(yùn)算后使用getSideOutput(OutputTag)。此時(shí)將會(huì)獲得鍵入旁路輸出流的結(jié)果。
Java:
final OutputTag outputTag = new OutputTag("side-output"){};
SingleOutputStreamOperator mainDataStream = ...;
DataStream sideOutputStream = mainDataStream.getSideOutput(outputTag);
Scala:
val outputTag = OutputTag[String]("side-output")
val mainDataStream = ...
val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)