Skip to content

Instantly share code, notes, and snippets.

@syhily

syhily/TwoStreamingJoining.scala

Last active July 6, 2018 18:36
Download ZIP
Raw
TwoStreamingJoining.scala
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, KeyedBroadcastProcessFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector
/**
* Joining and matching rules
*
* @author せいうはん (Employee ID: 17092068)
* @version 1.0.0, 2018-07-06 15:09
* @since 1.0.0, 2018-07-06 15:09
*/
object TwoStreamingJoining extends App {
private val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
private val rule: DataStream[(String, Int)] = environment
.fromCollection(
List(
("A", 1),
("B", 2),
("C", 3),
("D", 4),
("E", 5),
("F", 6),
("G", 7),
("H", 8),
("I", 9),
("J", 10),
("K", 11),
("L", 12),
("M", 13),
("N", 14),
("O", 15),
("AA", 16),
("BB", 17),
("CC", 18),
("DD", 19),
("EE", 20),
("FF", 21),
("GG", 22),
("HH", 23),
("II", 24),
("JJ", 25),
("KK", 26),
("LL", 27),
("MM", 28),
("NN", 29),
("OO", 30),
("AAA", 31),
("BBB", 32),
("CCC", 33),
("DDD", 34),
("EEE", 35),
("FFF", 36),
("GGG", 37),
("HHH", 38),
("III", 39),
("JJJ", 40),
("KKK", 41),
("LLL", 42),
("MMM", 43),
("NNN", 44),
("OOO", 45)
)
)
.setParallelism(1)
private val elements: DataStream[(String, Int)] = environment
.fromCollection(
List(
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1)
)
)
.setParallelism(1)
private val ruleBroadcastState =
new MapStateDescriptor("RuleBroadcast", createTypeInformation[String], createTypeInformation[(String, Int)])
private val broadcastStream: BroadcastStream[(String, Int)] = rule.broadcast(ruleBroadcastState)
rule.broadcast
.connect(elements)
.keyBy(a => a._1, b => b._1)
.flatMap(new CoFlatMapFunction[(String, Int), (String, Int), (String, Int)] {
private val sum = new scala.collection.mutable.HashMap[String, Int]
override def flatMap1(value: (String, Int), out: Collector[(String, Int)]): Unit = {
sum(value._1) = value._2
println(s"Listed: ${sum.keys}")
}
override def flatMap2(value: (String, Int), out: Collector[(String, Int)]): Unit =
if (sum.contains(value._1)) {
println(s"Passed: ${value._1}, ${value._2}, keys ${sum.keys}")
out.collect(value)
} else {
println(s"Skip: ${value._1} ${value._2}, keys ${sum.keys}")
}
})
.keyBy(0)
.connect(broadcastStream)
.process(new KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), Int] {
override def processElement(
value: (String, Int),
ctx: KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), Int]#ReadOnlyContext,
out: Collector[Int]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
if (state.contains(value._1)) {
out.collect(state.get(value._1)._2)
println(s"Matched rule: ${value._1}, ${value._2}, ${state.immutableEntries()}")
}
}
override def processBroadcastElement(
value: (String, Int),
ctx: KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), Int]#Context,
out: Collector[Int]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
state.put(value._1, value)
println(s"Update broadcast state: ${value._1}")
}
})
.print()
environment.setParallelism(1)
environment.execute()
}
Raw
TwoStreamingJoiningMultiplyTimes.scala
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, KeyedBroadcastProcessFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector
/**
* Joining and matching rules
*
* @author せいうはん (Employee ID: 17092068)
* @version 1.0.0, 2018-07-06 15:09
* @since 1.0.0, 2018-07-06 15:09
*/
object TwoStreamingJoiningMultiplyTimes extends App {
private val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
private val rule: DataStream[(String, Int)] = environment
.fromCollection(
List(
("A", 1),
("B", 2),
("C", 3),
("D", 4),
("E", 5),
("F", 6),
("G", 7),
("H", 8),
("I", 9),
("J", 10),
("K", 11),
("L", 12),
("M", 13),
("N", 14),
("O", 15),
("AA", 16),
("BB", 17),
("CC", 18),
("DD", 19),
("EE", 20),
("FF", 21),
("GG", 22),
("HH", 23),
("II", 24),
("JJ", 25),
("KK", 26),
("LL", 27),
("MM", 28),
("NN", 29),
("OO", 30),
("AAA", 31),
("BBB", 32),
("CCC", 33),
("DDD", 34),
("EEE", 35),
("FFF", 36),
("GGG", 37),
("HHH", 38),
("III", 39),
("JJJ", 40),
("KKK", 41),
("LLL", 42),
("MMM", 43),
("NNN", 44),
("OOO", 45)
)
)
.setParallelism(1)
private val elements: DataStream[(String, Int)] = environment
.fromCollection(
List(
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1),
("A", 6),
("A", 5),
("A", 4),
("A", 3),
("A", 2),
("A", 1),
("B", 6),
("B", 5),
("B", 4),
("B", 3),
("B", 2),
("B", 1),
("C", 6),
("C", 5),
("C", 4),
("C", 3),
("C", 2),
("C", 1),
("D", 6),
("D", 5),
("D", 4),
("D", 3),
("D", 2),
("D", 1)
)
)
.setParallelism(1)
private val ruleBroadcastState =
new MapStateDescriptor("RuleBroadcast", createTypeInformation[String], createTypeInformation[(String, Int)])
private val broadcastStream: BroadcastStream[(String, Int)] = rule.broadcast(ruleBroadcastState)
rule.broadcast
.connect(elements)
.keyBy(a => a._1, b => b._1)
.flatMap(new CoFlatMapFunction[(String, Int), (String, Int), (String, Int)] {
private val sum = new scala.collection.mutable.HashMap[String, Int]
override def flatMap1(value: (String, Int), out: Collector[(String, Int)]): Unit = {
sum(value._1) = value._2
println(s"Listed: ${sum.keys}")
}
override def flatMap2(value: (String, Int), out: Collector[(String, Int)]): Unit =
if (sum.contains(value._1)) {
println(s"Passed: ${value._1}, ${value._2}, keys ${sum.keys}")
out.collect(value)
} else {
println(s"Skip: ${value._1} ${value._2}, keys ${sum.keys}")
}
})
.keyBy(0)
.connect(broadcastStream)
.process(new KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), String] {
override def processElement(
value: (String, Int),
ctx: KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), String]#ReadOnlyContext,
out: Collector[String]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
if (state.contains(value._1)) {
out.collect(state.get(value._1)._1)
println(s"First Matched rule: ${value._1}, ${value._2}, ${state.immutableEntries()}")
}
}
override def processBroadcastElement(
value: (String, Int),
ctx: KeyedBroadcastProcessFunction[String, (String, Int), (String, Int), String]#Context,
out: Collector[String]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
state.put(value._1, value)
println(s"First Update broadcast state: ${value._1}")
}
})
.keyBy(s => s)
.connect(broadcastStream)
.process(new KeyedBroadcastProcessFunction[String, String, (String, Int), Int] {
override def processElement(
value: String,
ctx: KeyedBroadcastProcessFunction[String, String, (String, Int), Int]#ReadOnlyContext,
out: Collector[Int]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
if (state.contains(value)) {
out.collect(state.get(value)._2)
println(s"Second Matched rule: $value, ${state.immutableEntries()}")
}
}
override def processBroadcastElement(
value: (String, Int),
ctx: KeyedBroadcastProcessFunction[String, String, (String, Int), Int]#Context,
out: Collector[Int]
): Unit = {
val state = ctx.getBroadcastState(ruleBroadcastState)
state.put(value._1, value)
println(s"Second Update broadcast state: ${value._1}")
}
})
.print()
environment.setParallelism(1)
environment.execute()
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment