Apache Flink:ProcessWindowFunction KeyBy()多个值 [英] Apache Flink: ProcessWindowFunction KeyBy() multiple values
问题描述
我正在尝试将WindowFunction与DataStream一起使用，我的目标是拥有一个类似于以下内容的
I'm trying to use WindowFunction with DataStream, my goal is to have a Query like the following
SELECT *,
count(id) OVER(PARTITION BY country) AS c_country,
count(id) OVER(PARTITION BY city) AS c_city,
count(id) OVER(PARTITION BY city) AS c_addrs
FROM fm
ORDER BY country
已经帮助我按国家/地区字段进行汇总，但是我需要在同一时间窗口中按两个字段进行汇总. 我不知道在这种情况下 keyBy()中是否可以有两个或多个键
have helped me for the aggregation by the country field, but I need to do the aggregation by two fields in the same time window. I don't know if it is possible to have two or more keys in keyBy( ) for this case
val parsed = stream2.map(x=> {
val arr = x.split(",")
(arr(0).toInt, arr(1), arr(2))
})
parsed
.keyBy(x => x._2)
.window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.process(new ProcessWindowFunction[
(Int, String, String), (Int, String, String, Int), String, TimeWindow
]() {
override def process(key: String, context: Context,
elements: Iterable[(Int, String, String)],
out: Collector[(Int, String, String, Int)]): Unit = {
val lst = elements.toList
lst.foreach(x => out.collect((x._1, x._2, x._3, lst.size)))
}
}).print().setParallelism(1)
这对于第一次汇总非常有用，但是我在同一时间窗口中缺少城市字段的第二次汇总.
This is great for the first aggregation, but I'm missing the second aggregation for the city field in the same time window.
输入数据:
input data:
10,"SPAIN","BARCELONA","C1"
20,"SPAIN","BARCELONA","C2"
30,"SPAIN","MADRID","C3"
30,"SPAIN","MADRID","C3"
80,"SPAIN","MADRID","C4"
90,"SPAIN","VALENCIA","C5"
40,"ITALY","ROMA","C6"
41,"ITALY","ROMA","C7"
42,"ITALY","VENECIA","C8"
50,"FRANCE","PARIS","C9"
60,"FRANCE","PARIS","C9"
70,"FRANCE","MARSELLA","C10"
预期产量
expected output
(10,"SPAIN","BARCELONA",6,2,1)
(20,"SPAIN","BARCELONA",6,2,1)
(30,"SPAIN","MADRID",6,3,2)
(30,"SPAIN","MADRID",6,3,2)
(80,"SPAIN","MADRID",6,3,1)
(90,"SPAIN","VALENCIA",6,1,1)
(50,"FRANCE","PARIS",3,2,1)
(60,"FRANCE","PARIS",3,2,1)
(70,"FRANCE","MARSELLA",3,1,1)
(40,"ITALY","ROMA",3,2,2)
(41,"ITALY","ROMA",3,2,2)
(42,"ITALY","VENECIA",3,1,1)
---------------- 更新2 ------------------
---------------- UPDATE 2 ------------------
我目前想对3列进行汇总.如果我使用的选项是链接KeyBy()输出，但这会变得很长很复杂，而且可读性很差. 除此之外，我放置了一个Time.seconds(1)的时间窗口，因为如果没有此窗口，则上面的KeyBy()输出将作为单个事件.
I currently want to do the aggregation for 3 columns. If the option I'm using is to chain the KeyBy() output, but this can become very long and complex, and not very readable. In addition to this, I put a time window of Time.seconds(1) because without this window the KeyBy() output above takes as individual events.
我的兴趣是，是否可以在单个流程函数中进行这些聚合.
my interest is if I can make these aggregations in a single process function.
我的代码很长...
parsed
.keyBy(_.country) // key by product id.
.window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.process(new ProcessWindowFunction[
AlarmasIn, AlarmasOut, String, TimeWindow
]() {
override def process(key: String, context: Context,
elements: Iterable[AlarmasIn],
out: Collector[AlarmasOut]): Unit = {
val lst = elements.toList
lst.foreach(x => out.collect(AlarmasOut(x.id, x.country, x.city,x.address, lst.size,0,0)))
}
})
.keyBy( _.city).window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
.process(new ProcessWindowFunction[
AlarmasOut, AlarmasOut, String, TimeWindow
]() {
override def process(key: String,
context: Context,
elements: Iterable[AlarmasOut],
out: Collector[AlarmasOut]): Unit = {
val lst = elements.toList
lst.foreach(x => out.collect(AlarmasOut(x.id, x.country, x.city,x.address,x.c_country,lst.size,x.c_addr)))
}
})
.keyBy( _.address).window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
.process(new ProcessWindowFunction[
AlarmasOut, AlarmasOut, String, TimeWindow
]() {
override def process(key: String,
context: Context,
elements: Iterable[AlarmasOut],
out: Collector[AlarmasOut]): Unit = {
val lst = elements.toList
lst.foreach(x => out.collect(AlarmasOut(x.id, x.country, x.city,x.address,x.c_country,x.c_city,lst.size)))
}
})
.print()
/// CASE CLASS
case class AlarmasIn(
id: Int,
country: String,
city: String,
address: String
)
case class AlarmasOut(
id: Int,
country: String,
city: String,
address: String,
c_country: Int,
c_city: Int,
c_addr: Int
)
推荐答案
由于city是country的子类别，因此您可以先按city维度聚合流，然后再按country维度进行流聚合
As city is a subcategory of country, you can aggregate the stream by city dimension first, then do another aggregation by country dimension.
val parsed = stream2.map(x=> {
val arr = x.split(",")
(arr(0).toInt, arr(1), arr(2))
})
parsed
.keyBy(x => x._3)
.window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.process(new ProcessWindowFunction[
(Int, String, String), (Int, String, String, Int), String, TimeWindow
]() {
override def process(key: String, context: Context,
elements: Iterable[(Int, String, String)],
out: Collector[(Int, String, String, Int)]): Unit = {
val lst = elements.toList
lst.foreach(x => out.collect((x._1, x._2, x._3, lst.size)))
}
})
.keyBy(x => x._2)
.process(new ProcessWindowFunction[
(Int, String, String), (Int, String, String, Int), String, TimeWindow
]() {
override def process(key: String, context: Context,
elements: Iterable[(Int, String, String)],
out: Collector[(Int, String, String, Int)]): Unit = {
val cnt = 0
for(e:elements){
cnt += e._4
}
lst.foreach(x => out.collect((x._1, x._2, x._3, cnt)))
}
}).print().setParallelism(1)
如果一个维度不是另一个维度的子维度，则可以合并这2个维度并生成新的键，然后由您自己在流程函数中实现聚合逻辑.
If one dimension is not a sub-dimension of the other dim, you can concat these 2 dims and generete a new key, then implement the aggregation logic in process func by yourself.
keyBy(x=>x._2+x._3)
更新
我认为无法在您尝试使用不同键进行统计的一个过程函数cuz中计算结果.一步执行此操作的唯一方法是将全局并行性设置为1(即使使用keyby func，所有输入数据也将进入一个下游任务)或将输入数据广播到所有下游任务.
UPDATE
I think it is not possible to calculate the result in one process function cuz you are trying to do the statistics with different keys. The only way to do it in one step is that you set the global parallelism to 1(all input data will go to one downstream task even you use a keyby func) or broadcast the input data to all downstream tasks.
由于您的计算实际上具有一些通用的过程逻辑，因此最好进行一些抽象.
Since your calculation actually have some common process logic, it would be better to do some abstraction.
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
object CountJob {
@throws[Exception]
def main(args: Array[String]): Unit = {
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val transactions: DataStream[Record] = env
.addSource(new SourceFunction[Record] {
override def run(sourceContext: SourceFunction.SourceContext[Record]): Unit = {
while (true) {
sourceContext.collect(Record(1, "a", "b", "c", 1, 1, 1))
Thread.sleep(1000)
}
}
override def cancel(): Unit = {
}
})
.name("generate source")
transactions.keyBy(_.addr)
.timeWindow(Time.seconds(1))
.process(new CustomCountProc("ADDR"))
.keyBy(_.city)
.timeWindow(Time.seconds(1))
.process(new CustomCountProc("CITY"))
.keyBy(_.country)
.timeWindow(Time.seconds(1))
.process(new CustomCountProc("COUNTRY"))
.print()
env.execute("Count Job")
}
}
// a common operator to process different aggregation
class CustomCountProc(aggrType: String) extends ProcessWindowFunction[Record, Record, String, TimeWindow] {
override def process(key: String, context: Context, elements: Iterable[Record], out: Collector[Record]): Unit = {
for (e <- elements) {
if ("ADDR".equals(aggrType)) {
out.collect(Record(-1, e.country, e.city, key, e.country_cnt, e.city_cnt, elements.size))
}
else if ("CITY".equals(aggrType)) {
out.collect(Record(-1, e.country, key, e.country, e.country_cnt, elements.size, e.addr_cnt))
}
else if ("COUNTRY".equals(aggrType)) {
out.collect(Record(-1, key, e.city, e.addr, elements.size, e.city_cnt, e.addr_cnt))
}
}
}
}
case class Record(
id: Int,
country: String,
city: String,
addr: String,
country_cnt: Int,
city_cnt: Int,
addr_cnt: Int
) {
}
顺便说一句，我不确定输出是否确实符合您的期望.由于您未实现有状态流程功能，因此我认为您正在尝试计算每一批数据的聚合结果，并且每一批都包含在一秒钟的时间窗口内提取的数据.输出不会一直累积，每个批次将从零开始.
Btw, I am not sure if the output actually meets your expectation. As you didn't implement a stateful process function, I think you are trying to calculate the aggregation results for each batch of data, and each batch contains data ingested in a time window of one second. The output won't accumulate all the time, each batch will start from zero.
通过使用timeWindow功能，您还需要注意TimeCharacteristic，默认情况下是处理时间.
By using timeWindow function, you also need to notice the TimeCharacteristic which by default is the processing time.
由于使用了3个随后的window功能，输出可能也会延迟.假设第一个流程函子在一秒钟内完成了聚合并将结果转发到下游.由于第二个进程func的timewindow也为1秒，因此它将不会发出任何结果，直到它从上游接收到下一批输出为止.
The output may also delay because of using 3 consequent window function. Suppose the first process func has completed the aggregation within one second and forward the results downstream. As the second process func also has a timewindow of 1 second, it won't emit any result until it receives next batch of output from upstream.
让我们看看其他人是否可以更好地解决您的问题.
Let's see if others have better solution to your problem.
这篇关于Apache Flink:ProcessWindowFunction KeyBy()多个值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
