聊聊flink Table的Group Windows
摘要：
序
本文主要研究一下flink Table的Group Windows
例項
Table table = input
.window([Window w].as("w"))// define window with alias w
.groupBy(&#...
序
本文主要研究一下flink Table的Group Windows
例項
Table table = input
.window([Window w].as("w"))// define window with alias w
.groupBy("w")// group the table by window w
.select("b.sum");// aggregate
Table table = input
.window([Window w].as("w"))// define window with alias w
.groupBy("w, a")// group the table by attribute a and window w
.select("a, b.sum");// aggregate
Table table = input
.window([Window w].as("w"))// define window with alias w
.groupBy("w, a")// group the table by attribute a and window w
.select("a, w.start, w.end, w.rowtime, b.count"); // aggregate and add window start, end, and rowtime timestamps
- window操作可以對Window進行別名，然後可以在groupBy及select中引用，window有start、end、rowtime屬性可以用，其中start及rowtime是inclusive的，而end為exclusive
Tumbling Windows例項
// Tumbling Event-time Window
.window(Tumble.over("10.minutes").on("rowtime").as("w"));
// Tumbling Processing-time Window (assuming a processing-time attribute "proctime")
.window(Tumble.over("10.minutes").on("proctime").as("w"));
// Tumbling Row-count Window (assuming a processing-time attribute "proctime")
.window(Tumble.over("10.rows").on("proctime").as("w"));
- Tumbling Windows按固定視窗大小來移動，因而視窗不重疊；over方法用於指定視窗大小；視窗大小可以基於event-time、processing-time、row-count來定義
Sliding Windows例項
// Sliding Event-time Window
.window(Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w"));
// Sliding Processing-time window (assuming a processing-time attribute "proctime")
.window(Slide.over("10.minutes").every("5.minutes").on("proctime").as("w"));
// Sliding Row-count window (assuming a processing-time attribute "proctime")
.window(Slide.over("10.rows").every("5.rows").on("proctime").as("w"));
- Sliding Windows在slide interval小於window size的時候，視窗會有重疊，因而rows可能歸屬多個視窗；over方法用於指定視窗大小，視窗大小可以基於event-time、processing-time、row-count來定義；every方法用於指定slide interval
Session Windows例項
// Session Event-time Window
.window(Session.withGap("10.minutes").on("rowtime").as("w"));
// Session Processing-time Window (assuming a processing-time attribute "proctime")
.window(Session.withGap("10.minutes").on("proctime").as("w"));
- Session Windows沒有固定的視窗大小，它基於inactivity的程度來關閉視窗，withGap方法用於指定兩個視窗的gap，作為time interval；Session Windows只能使用event-time或者processing-time
Table.window
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala
class Table(
private[flink] val tableEnv: TableEnvironment,
private[flink] val logicalPlan: LogicalNode) {
//......
def window(window: Window): WindowedTable = {
new WindowedTable(this, window)
}
//......
}
- Table提供了window操作，接收Window引數，建立的是WindowedTable
WindowedTable
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala
class WindowedTable(
private[flink] val table: Table,
private[flink] val window: Window) {
def groupBy(fields: Expression*): WindowGroupedTable = {
val fieldsWithoutWindow = fields.filterNot(window.alias.equals(_))
if (fields.size != fieldsWithoutWindow.size + 1) {
throw new ValidationException("GroupBy must contain exactly one window alias.")
}
new WindowGroupedTable(table, fieldsWithoutWindow, window)
}
def groupBy(fields: String): WindowGroupedTable = {
val fieldsExpr = ExpressionParser.parseExpressionList(fields)
groupBy(fieldsExpr: _*)
}
}
- WindowedTable只提供groupBy操作，其中groupBy可以接收String型別的引數，也可以接收Expression型別的引數；String型別的引數會被轉換為Expression型別，最後呼叫的是Expression型別引數的groupBy方法；如果groupBy除了window沒有其他屬性，則其parallelism為1，只會在單一task上執行；groupBy方法建立的是WindowGroupedTable
WindowGroupedTable
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala
class WindowGroupedTable(
private[flink] val table: Table,
private[flink] val groupKeys: Seq[Expression],
private[flink] val window: Window) {
def select(fields: Expression*): Table = {
val expandedFields = expandProjectList(fields, table.logicalPlan, table.tableEnv)
val (aggNames, propNames) = extractAggregationsAndProperties(expandedFields, table.tableEnv)
val projectsOnAgg = replaceAggregationsAndProperties(
expandedFields, table.tableEnv, aggNames, propNames)
val projectFields = extractFieldReferences(expandedFields ++ groupKeys :+ window.timeField)
new Table(table.tableEnv,
Project(
projectsOnAgg,
WindowAggregate(
groupKeys,
window.toLogicalWindow,
propNames.map(a => Alias(a._1, a._2)).toSeq,
aggNames.map(a => Alias(a._1, a._2)).toSeq,
Project(projectFields, table.logicalPlan).validate(table.tableEnv)
).validate(table.tableEnv),
// required for proper resolution of the time attribute in multi-windows
explicitAlias = true
).validate(table.tableEnv))
}
def select(fields: String): Table = {
val fieldExprs = ExpressionParser.parseExpressionList(fields)
//get the correct expression for AggFunctionCall
val withResolvedAggFunctionCall = fieldExprs.map(replaceAggFunctionCall(_, table.tableEnv))
select(withResolvedAggFunctionCall: _*)
}
}
- WindowGroupedTable只提供select操作，其中select可以接收String型別的引數，也可以接收Expression型別的引數；String型別的引數會被轉換為Expression型別，最後呼叫的是Expression型別引數的select方法；select方法建立了新的Table，其Project的child為WindowAggregate
小結
- window操作可以對Window進行別名，然後可以在groupBy及select中引用，window有start、end、rowtime屬性可以用，其中start及rowtime是inclusive的，而end為exclusive
- Tumbling Windows按固定視窗大小來移動，因而視窗不重疊；over方法用於指定視窗大小；視窗大小可以基於event-time、processing-time、row-count來定義；Sliding Windows在slide interval小於window size的時候，視窗會有重疊，因而rows可能歸屬多個視窗；over方法用於指定視窗大小，視窗大小可以基於event-time、processing-time、row-count來定義；every方法用於指定slide interval；Session Windows沒有固定的視窗大小，它基於inactivity的程度來關閉視窗，withGap方法用於指定兩個視窗的gap，作為time interval；Session Windows只能使用event-time或者processing-time
- Table提供了window操作，接收Window引數，建立的是WindowedTable；WindowedTable只提供groupBy操作，其中groupBy可以接收String型別的引數，也可以接收Expression型別的引數；String型別的引數會被轉換為Expression型別，最後呼叫的是Expression型別引數的groupBy方法；如果groupBy除了window沒有其他屬性，則其parallelism為1，只會在單一task上執行；groupBy方法建立的是WindowGroupedTable；WindowGroupedTable只提供select操作，其中select可以接收String型別的引數，也可以接收Expression型別的引數；String型別的引數會被轉換為Expression型別，最後呼叫的是Expression型別引數的select方法；select方法建立了新的Table，其Project的child為WindowAggregate