java消息顺序执行_Apache Flink：如何并行执行但保持消息顺序？
請(qǐng)?jiān)谙旅嬲业绞褂脗?cè)輸出和插槽組進(jìn)行本地?cái)U(kuò)展的示例 .
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")
}
}
總結(jié)
以上是生活随笔為你收集整理的java消息顺序执行_Apache Flink：如何并行执行但保持消息顺序？的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Nginx面试题总结大全
- 下一篇： 实验5 编写、调试具有多个段的程序