請在下面找到使用側輸出和插槽組進行本地擴展的示例 .

package org.example

/*

* Licensed to the Apache Software Foundation (ASF) under one

* or more contributor license agreements. See the NOTICE file

* distributed with this work for additional information

* regarding copyright ownership. The ASF licenses this file

* to you under the Apache License, Version 2.0 (the

* "License"); you may not use this file except in compliance

* with the License. You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

import org.apache.flink.streaming.api.functions.ProcessFunction

import org.apache.flink.streaming.api.scala._

import org.apache.flink.util.Collector

/**

* This example shows an implementation of WordCount with data from a text socket.

* To run the example make sure that the service providing the text data is already up and running.

*

* To start an example socket text stream on your local machine run netcat from a command line,

* where the parameter specifies the port number:

*

* {{{

* nc -lk 9999

* }}}

*

* Usage:

* {{{

* SocketTextStreamWordCount

* }}}

*

* This example shows how to:

*

* - use StreamExecutionEnvironment.socketTextStream

* - write a simple Flink Streaming program in scala.

* - write and use user-defined functions.

*/

object SocketTextStreamWordCount {

def main(args: Array[String]) {

if (args.length != 2) {

System.err.println("USAGE:\nSocketTextStreamWordCount ")

return

}

val hostName = args(0)

val port = args(1).toInt

val outputTag1 = OutputTag[String]("side-1")

val outputTag2 = OutputTag[String]("side-2")

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.getConfig.enableObjectReuse()

//Create streams for names and ages by mapping the inputs to the corresponding objects

val text = env.socketTextStream(hostName, port).slotSharingGroup("processElement")

val counts = text.flatMap {

_.toLowerCase.split("\\W+") filter {

_.nonEmpty

}

}

.process(new ProcessFunction[String, String] {

override def processElement(

value: String,

ctx: ProcessFunction[String, String]#Context,

out: Collector[String]): Unit = {

if (value.head <= 'm') ctx.output(outputTag1, String.valueOf(value))

else ctx.output(outputTag2, String.valueOf(value))

}

})

val sideOutputStream1: DataStream[String] = counts.getSideOutput(outputTag1)

val sideOutputStream2: DataStream[String] = counts.getSideOutput(outputTag2)

val output1 = sideOutputStream1.map {

(_, 1)

}.slotSharingGroup("map1")

.keyBy(0)

.sum(1)

val output2 = sideOutputStream2.map {

(_, 1)

}.slotSharingGroup("map2")

.keyBy(0)

.sum(1)

output1.print()

output2.print()

env.execute("Scala SocketTextStreamWordCount Example")

}

}