Skip to content

Instantly share code, notes, and snippets.

@syhily

syhily/TwoStreamingJoining.scala

Last active July 6, 2018 18:36
Show Gist options
  • Save syhily/932e0d1e0f12b3e951236d6c36e5a7ed to your computer and use it in GitHub Desktop.
Save syhily/932e0d1e0f12b3e951236d6c36e5a7ed to your computer and use it in GitHub Desktop.
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