FlinkTableApi详解（算子）

该文章主要是对Flink官网相关内容进行翻译&＃xff0c;原文地址&＃xff1a;https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/tableApi.html#over-windows

Table API是用于流和批处理的统一的关系API。Table API查询可以在批处理或流输入上运行而无需修改。Table API是SQL语言的超集&＃xff0c;是专门为使用Apache Flink设计的。Table API是用于Scala和Java的语言集成的API。 Table API查询不是像SQL中常见的那样将查询指定为String值&＃xff0c;而是以Java或Scala中的语言嵌入样式定义&＃xff0c;并具有IDE支持&＃xff0c;例如自动完成和语法验证。

Table API与Flink的SQL集成共享其API的许多概念和部分。看一下Common Concepts&＃xff06;API&＃xff0c;了解如何注册表或创建Table对象。“Streaming Concepts”页面讨论了流的特定概念&＃xff0c;例如动态表和时间属性。下面的示例假定具有属性(a, b, c, rowtime)的称为Orders的已注册表。rowtime字段是流中的逻辑时间属性&＃xff0c;或者是批处理中的常规时间戳记字段。

概述与示例

Table API可用于Scala和Java。 Scala Table API利用Scala表达式&＃xff0c;Java Table API基于字符串&＃xff0c;这些字符串被解析并转换为等效表达式。

以下示例显示了Scala和Java Table API之间的区别。该Table程序在批处理环境中执行。它将扫描“Orders”表&＃xff0c;按字段a进行分组&＃xff0c;并计算每组的结果行。该Table程序的结果将转换为Row类型的DataSet并进行打印。

Java

通过导入org.apache.flink.table.api.java._启用Java Table API。以下示例显示了如何构造Java Table API程序以及如何将表达式指定为字符串。

import org.apache.flink.table.api._
import org.apache.flink.table.api.java._// environment configuration
ExecutionEnvironment env &＃61; ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv &＃61; BatchTableEnvironment.create(env);// register Orders table in table environment
// ...// specify table program
Table orders &＃61; tEnv.from("Orders"); // schema (a, b, c, rowtime)Table counts &＃61; orders.groupBy("a").select("a, b.count as cnt");// conversion to DataSet
DataSet<Row> result &＃61; tEnv.toDataSet(counts, Row.class);
result.print();


Scala

通过导入org.apache.flink.api.scala._和org.apache.flink.table.api.scala._来启用Scala Table API。以下示例显示了Scala Table API程序的构造方式。使用Scala符号引用Table属性&＃xff0c;Scala符号以撇号&＃xff08;’&＃xff09;开头。

import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._// environment configuration
val env &＃61; ExecutionEnvironment.getExecutionEnvironment
val tEnv &＃61; BatchTableEnvironment.create(env)// register Orders table in table environment
// ...// specify table program
val orders &＃61; tEnv.from("Orders") // schema (a, b, c, rowtime)val result &＃61; orders.groupBy(&＃39;a).select(&＃39;a, &＃39;b.count as &＃39;cnt).toDataSet[Row] // conversion to DataSet.print()


下一个示例显示了一个更复杂的Table API程序。程序再次扫描“Orders”表。它过滤空值&＃xff0c;对String类型的字段a进行归一化&＃xff0c;并针对每个小时计算并产生a平均帐单金额b。

Java

// environment configuration
// ...// specify table program
Table orders &＃61; tEnv.from("Orders"); // schema (a, b, c, rowtime)Table result &＃61; orders.filter("a.isNotNull && b.isNotNull && c.isNotNull").select("a.lowerCase() as a, b, rowtime").window(Tumble.over("1.hour").on("rowtime").as("hourlyWindow")).groupBy("hourlyWindow, a").select("a, hourlyWindow.end as hour, b.avg as avgBillingAmount");


Scala

// environment configuration
// ...// specify table program
val orders: Table &＃61; tEnv.from("Orders") // schema (a, b, c, rowtime)val result: Table &＃61; orders.filter(&＃39;a.isNotNull && &＃39;b.isNotNull && &＃39;c.isNotNull).select(&＃39;a.lowerCase() as &＃39;a, &＃39;b, &＃39;rowtime).window(Tumble over 1.hour on &＃39;rowtime as &＃39;hourlyWindow).groupBy(&＃39;hourlyWindow, &＃39;a).select(&＃39;a, &＃39;hourlyWindow.end as &＃39;hour, &＃39;b.avg as &＃39;avgBillingAmount)


由于Table API是用于批处理和流数据的统一API&＃xff0c;因此两个示例程序都可以在批处理和流输入上执行&＃xff0c;而无需对Table程序本身进行任何修改。在这两种情况下&＃xff0c;只要流记录不晚&＃xff0c;程序都会产生相同的结果。

算子

Table API支持以下算子操作。 请注意&＃xff0c;并非所有操作都可用于批处理和流式处理; 他们被相应地标记。

Scan, Projection, and Filter

Column Operations

Aggregations

Joins

集合算子

OrderBy, Offset & Fetch

Insert

Group Windows

组窗口根据时间或行计数&＃xff08;row-count &＃xff09;间隔将行组聚合为有限组&＃xff0c;并按组聚合函数。 对于批处理表&＃xff0c;窗口是按时间间隔对记录进行分组的便捷快捷方式。

Windows是使用window&＃xff08;w&＃xff1a;GroupWindow&＃xff09;子句定义的&＃xff0c;并且需要使用as子句指定的别名。为了按窗口对表进行分组&＃xff0c;必须像常规分组属性一样在groupBy&＃xff08;…&＃xff09;子句中引用窗口别名。

以下示例显示如何在表上定义窗口聚合。

val table &＃61; input.window([w: GroupWindow] as &＃39;w) // define window with alias w.groupBy(&＃39;w) // group the table by window w.select(&＃39;b.sum) // aggregate


在流式环境中&＃xff0c;如果窗口聚合除了窗口之外还在一个或多个属性上进行分组&＃xff0c;则它们只能并行计算。即groupBy&＃xff08;…&＃xff09;子句引用了窗口别名和至少一个其他属性。仅引用窗口别名的groupBy&＃xff08;…&＃xff09;子句&＃xff08;例如上例中的子句&＃xff09;只能由单个非并行任务求值。以下示例显示如何使用其他分组属性定义窗口聚合。

val table &＃61; input.window([w: GroupWindow] as &＃39;w) // define window with alias w.groupBy(&＃39;w, &＃39;a) // group the table by attribute a and window w .select(&＃39;a, &＃39;b.sum) // aggregate


可以在select语句中将窗口属性&＃xff08;例如时间窗口的开始&＃xff0c;结束或行时间戳&＃xff09;添加为窗口别名的属性&＃xff0c;分别为w.start&＃xff0c;w.end和w.rowtime。窗口开始和行时间时间戳是包含窗口的上下边界。相反&＃xff0c;窗口结束时间戳是唯一的窗口上边界。例如&＃xff0c;从下午2点开始的30分钟滚动窗口将以14&＃xff1a;00&＃xff1a;00.000作为开始时间戳&＃xff0c;以14&＃xff1a;29&＃xff1a;59.999作为行时间时间戳&＃xff0c;以14&＃xff1a;30&＃xff1a;00.000作为结束时间戳。

val table &＃61; input.window([w: GroupWindow] as &＃39;w) // define window with alias w.groupBy(&＃39;w, &＃39;a) // group the table by attribute a and window w .select(&＃39;a, &＃39;w.start, &＃39;w.end, &＃39;w.rowtime, &＃39;b.count) // aggregate and add window start, end, and rowtime timestamps


Window参数定义如何将行映射到窗口。窗口不是用户可以实现的接口。相反&＃xff0c;Table API提供了一组具有特定语义的预定义Window类&＃xff0c;这些类被转换为基础的DataStream或DataSet操作。支持的窗口定义在下面列出。

Tumble (滚动窗口)

滚动窗口将行分配给固定长度的不重叠的连续窗口。

例如&＃xff0c;5分钟的翻滚窗口以5分钟为间隔对行进行分组。 可以在事件时间&＃xff0c;处理时间或行数上定义翻滚窗口。使用Tumble类定义翻滚窗口如下&＃xff1a;

滚动窗口是使用Tumble类定义的&＃xff0c;如下所示&＃xff1a;

/ Tumbling Event-time Window
.window(Tumble over 10.minutes on &＃39;rowtime as &＃39;w)// Tumbling Processing-time Window (assuming a processing-time attribute "proctime")
.window(Tumble over 10.minutes on &＃39;proctime as &＃39;w)// Tumbling Row-count Window (assuming a processing-time attribute "proctime")
.window(Tumble over 10.rows on &＃39;proctime as &＃39;w)


Slide (滑动窗口)

滑动窗口的大小固定&＃xff0c;并以指定的滑动间隔滑动。如果滑动间隔小于窗口大小&＃xff0c;则滑动窗口重叠。因此&＃xff0c;可以将行分配给多个窗口。

例如&＃xff0c;15分钟大小和5分钟滑动间隔的滑动窗口将每行分配给3个不同的15分钟大小的窗口&＃xff0c;这些窗口以5分钟的间隔进行调用。 可以在事件时间&＃xff0c;处理时间或行数上定义滑动窗口。

滑动窗口是通过使用Slide类定义的&＃xff0c;如下所示&＃xff1a;

// Sliding Event-time Window
.window(Slide over 10.minutes every 5.minutes on &＃39;rowtime as &＃39;w)// Sliding Processing-time window (assuming a processing-time attribute "proctime")
.window(Slide over 10.minutes every 5.minutes on &＃39;proctime as &＃39;w)// Sliding Row-count window (assuming a processing-time attribute "proctime")
.window(Slide over 10.rows every 5.rows on &＃39;proctime as &＃39;w)


Session (会话窗口)

会话窗口没有固定的大小&＃xff0c;但是其边界由不活动的时间间隔定义&＃xff0c;即如果在定义的间隔时间内没有事件出现&＃xff0c;则会话窗口关闭。

例如&＃xff0c;间隔30分钟的会话窗口会在30分钟不活动后观察到一行&＃xff08;否则该行将被添加到现有窗口&＃xff09;后开始&＃xff0c;如果30分钟内未添加任何行&＃xff0c;则关闭该窗口。会话窗口可以在事件时间或处理时间工作。

会话窗口是通过使用Session类定义的&＃xff0c;如下所示&＃xff1a;

// Session Event-time Window
.window(Session withGap 10.minutes on &＃39;rowtime as &＃39;w)// Session Processing-time Window (assuming a processing-time attribute "proctime")
.window(Session withGap 10.minutes on &＃39;proctime as &＃39;w)


Over Windows

窗口聚合是标准SQL&＃xff08;OVER子句&＃xff09;已知的&＃xff0c;并在查询的SELECT子句中定义。与在GROUP BY子句中指定的组窗口不同&＃xff0c;在窗口上方不会折叠行。取而代之的是在窗口聚合中&＃xff0c;为每个输入行在其相邻行的范围内计算聚合。

使用window&＃xff08;w&＃xff1a;OverWindow &＃xff09;子句&＃xff08;在Python API中使用over_window&＃xff08; OverWindow&＃xff09;&＃xff09;定义窗口&＃xff0c;并在select() 方法中通过别名引用。以下示例显示如何在表上定义窗口聚合。

val table &＃61; input.window([w: OverWindow] as &＃39;w) // define over window with alias w.select(&＃39;a, &＃39;b.sum over &＃39;w, &＃39;c.min over &＃39;w) // aggregate over the over window w


OverWindow定义了计算聚合的行范围。OverWindow不是用户可以实现的接口。相反&＃xff0c;Table API提供了Over类来配置over窗口的属性。可以在事件时间或处理时间以及指定为时间间隔或行计数的范围上定义窗口上方。受支持的over窗口定义作为Over&＃xff08;和其他类&＃xff09;上的方法公开&＃xff0c;并在下面列出&＃xff1a;

注意&＃xff1a;当前&＃xff0c;同一select&＃xff08;&＃xff09;调用中的所有聚合函数必须在相同的窗口范围内计算。

Unbounded Over Windows

// Unbounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy &＃39;a orderBy &＃39;rowtime preceding UNBOUNDED_RANGE as &＃39;w)// Unbounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy &＃39;a orderBy &＃39;proctime preceding UNBOUNDED_RANGE as &＃39;w)// Unbounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy &＃39;a orderBy &＃39;rowtime preceding UNBOUNDED_ROW as &＃39;w)// Unbounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy &＃39;a orderBy &＃39;proctime preceding UNBOUNDED_ROW as &＃39;w)


Bounded Over Windows

// Bounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy &＃39;a orderBy &＃39;rowtime preceding 1.minutes as &＃39;w)// Bounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy &＃39;a orderBy &＃39;proctime preceding 1.minutes as &＃39;w)// Bounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy &＃39;a orderBy &＃39;rowtime preceding 10.rows as &＃39;w)// Bounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy &＃39;a orderBy &＃39;proctime preceding 10.rows as &＃39;w)


基于行的操作

基于行的操作生成具有多列的输出。

• Map Batch Streaming

  使用用户定义的标量函数或内置标量函数执行映射操作。如果输出类型是复合类型&＃xff0c;则输出将被展平。

  class MyMapFunction extends ScalarFunction {def eval(a: String): Row &＃61; {Row.of(a, "pre-" &＃43; a)}override def getResultType(signature: Array[Class[_]]): TypeInformation[_] &＃61;Types.ROW(Types.STRING, Types.STRING)
}val func &＃61; new MyMapFunction()
val table &＃61; input.map(func(&＃39;c)).as(&＃39;a, &＃39;b)


• FlatMap Batch Streaming

  使用表函数执行flatMap操作。

  class MyFlatMapFunction extends TableFunction[Row] {def eval(str: String): Unit &＃61; {if (str.contains("#")) {str.split("#").foreach({ s &＃61;>val row &＃61; new Row(2)row.setField(0, s)row.setField(1, s.length)collect(row)})}}override def getResultType: TypeInformation[Row] &＃61; {Types.ROW(Types.STRING, Types.INT)}
}val func &＃61; new MyFlatMapFunction
val table &＃61; input.flatMap(func(&＃39;c)).as(&＃39;a, &＃39;b)


• Aggregate Batch Streaming Result Updating

  使用聚合函数执行聚合操作。您必须使用select语句关闭“聚合”&＃xff0c;并且select语句不支持聚合函数。如果输出类型是复合类型&＃xff0c;则聚合的输出将被展平。

  case class MyMinMaxAcc(var min: Int, var max: Int)class MyMinMax extends AggregateFunction[Row, MyMinMaxAcc] {def accumulate(acc: MyMinMaxAcc, value: Int): Unit &＃61; {if (value acc.max) {acc.max &＃61; value}}override def createAccumulator(): MyMinMaxAcc &＃61; MyMinMaxAcc(0, 0)def resetAccumulator(acc: MyMinMaxAcc): Unit &＃61; {acc.min &＃61; 0acc.max &＃61; 0}override def getValue(acc: MyMinMaxAcc): Row &＃61; {Row.of(Integer.valueOf(acc.min), Integer.valueOf(acc.max))}override def getResultType: TypeInformation[Row] &＃61; {new RowTypeInfo(Types.INT, Types.INT)}
}val myAggFunc &＃61; new MyMinMax
val table &＃61; input.groupBy(&＃39;key).aggregate(myAggFunc(&＃39;a) as (&＃39;x, &＃39;y)).select(&＃39;key, &＃39;x, &＃39;y)


• Group Window Aggregate Batch Streaming

  在组窗口和可能的一个或多个分组键上对表进行分组和聚集。您必须使用select语句关闭“聚合”。并且select语句不支持“ *”或聚合函数。

  val myAggFunc &＃61; new MyMinMax
val table &＃61; input.window(Tumble over 5.minutes on &＃39;rowtime as &＃39;w) // define window.groupBy(&＃39;key, &＃39;w) // group by key and window.aggregate(myAggFunc(&＃39;a) as (&＃39;x, &＃39;y)).select(&＃39;key, &＃39;x, &＃39;y, &＃39;w.start, &＃39;w.end) // access window properties and aggregate results


• FlatAggregate Batch Streaming

  类似于GroupBy聚合。使用以下运行表聚合运算符将分组键上的行分组&＃xff0c;以逐行聚合行。与AggregateFunction的区别在于TableAggregateFunction可以为一个组返回0个或更多记录。您必须使用select语句关闭“ flatAggregate”。并且select语句不支持聚合函数。除了使用emitValue来输出结果之外&＃xff0c;还可以使用emitUpdateWithRetract方法。与emitValue不同&＃xff0c;emitUpdateWithRetract用于发出已更新的值。此方法在撤消模式下增量输出数据&＃xff0c;即&＃xff0c;一旦有更新&＃xff0c;我们就必须撤消旧记录&＃xff0c;然后再发送新的更新记录。如果在表聚合函数中定义了这两种方法&＃xff0c;则将优先使用emitUpdateWithRetract方法&＃xff0c;因为这两种方法比emitValue更有效&＃xff0c;因为它可以增量输出值。

  import java.lang.{Integer &＃61;> JInteger}
import org.apache.flink.table.api.Types
import org.apache.flink.table.functions.TableAggregateFunction/*** Accumulator for top2.*/
class Top2Accum {var first: JInteger &＃61; _var second: JInteger &＃61; _
}/*** The top2 user-defined table aggregate function.*/
class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {override def createAccumulator(): Top2Accum &＃61; {val acc &＃61; new Top2Accumacc.first &＃61; Int.MinValueacc.second &＃61; Int.MinValueacc}def accumulate(acc: Top2Accum, v: Int) {if (v > acc.first) {acc.second &＃61; acc.firstacc.first &＃61; v} else if (v > acc.second) {acc.second &＃61; v}}def merge(acc: Top2Accum, its: JIterable[Top2Accum]): Unit &＃61; {val iter &＃61; its.iterator()while (iter.hasNext) {val top2 &＃61; iter.next()accumulate(acc, top2.first)accumulate(acc, top2.second)}}def emitValue(acc: Top2Accum, out: Collector[JTuple2[JInteger, JInteger]]): Unit &＃61; {// emit the value and rankif (acc.first !&＃61; Int.MinValue) {out.collect(JTuple2.of(acc.first, 1))}if (acc.second !&＃61; Int.MinValue) {out.collect(JTuple2.of(acc.second, 2))}}
}val top2 &＃61; new Top2
val orders: Table &＃61; tableEnv.from("Orders")
val result &＃61; orders.groupBy(&＃39;key).flatAggregate(top2(&＃39;a) as (&＃39;v, &＃39;rank)).select(&＃39;key, &＃39;v, &＃39;rank)


  **Note:**对于流查询&＃xff0c;根据聚合的类型和不同的分组键的数量&＃xff0c;计算查询结果所需的状态可能会无限增长。请提供具有有效保留间隔的查询配置&＃xff0c;以防止出现过多的状态。有关详细信息&＃xff0c;请参见查询配置。

• Group Window FlatAggregate Batch Streaming

  在组窗口和可能的一个或多个分组键上对表进行分组和聚集。您必须使用select语句关闭“ flatAggregate”。并且select语句不支持聚合函数。

  val top2 &＃61; new Top2
val orders: Table &＃61; tableEnv.from("Orders")
val result &＃61; orders.window(Tumble over 5.minutes on &＃39;rowtime as &＃39;w) // define window.groupBy(&＃39;a, &＃39;w) // group by key and window.flatAggregate(top2(&＃39;b) as (&＃39;v, &＃39;rank)).select(&＃39;a, w.start, &＃39;w.end, &＃39;w.rowtime, &＃39;v, &＃39;rank) // access window properties and aggregate results


Data Types

请参阅有关数据类型的专用页面。通用类型和&＃xff08;嵌套的&＃xff09;复合类型&＃xff08;例如POJO&＃xff0c;元组&＃xff0c;行&＃xff0c;Scala案例类&＃xff09;也可以是一行的字段。可以使用值访问功能访问具有任意嵌套的复合类型的字段。泛型类型被视为黑盒&＃xff0c;可以通过用户定义的函数传递或处理。

表达式语法

前面几节中的某些运算符期望一个或多个表达式。可以使用嵌入式Scala DSL或字符串指定表达式。请参考上面的示例以了解如何指定表达式。

这是用于表达式的EBNF语法&＃xff1a;

expressionList &＃61; expression , { "," , expression } ;expression &＃61; overConstant | alias ;alias &＃61; logic | ( logic , "as" , fieldReference ) | ( logic , "as" , "(" , fieldReference , { "," , fieldReference } , ")" ) ;logic &＃61; comparison , [ ( "&&" | "||" ) , comparison ] ;comparison &＃61; term , [ ( "&＃61;" | "&＃61;&＃61;" | "&＃61;&＃61;&＃61;" | "!&＃61;" | "!&＃61;&＃61;" | ">" | ">&＃61;" | "<" | "<&＃61;" ) , term ] ;term &＃61; product , [ ( "&＃43;" | "-" ) , product ] ;product &＃61; unary , [ ( "*" | "/" | "%") , unary ] ;unary &＃61; [ "!" | "-" | "&＃43;" ] , composite ;composite &＃61; over | suffixed | nullLiteral | prefixed | atom ;suffixed &＃61; interval | suffixAs | suffixCast | suffixIf | suffixDistinct | suffixFunctionCall | timeIndicator ;prefixed &＃61; prefixAs | prefixCast | prefixIf | prefixDistinct | prefixFunctionCall ;interval &＃61; timeInterval | rowInterval ;timeInterval &＃61; composite , "." , ("year" | "years" | "quarter" | "quarters" | "month" | "months" | "week" | "weeks" | "day" | "days" | "hour" | "hours" | "minute" | "minutes" | "second" | "seconds" | "milli" | "millis") ;rowInterval &＃61; composite , "." , "rows" ;suffixCast &＃61; composite , ".cast(" , dataType , ")" ;prefixCast &＃61; "cast(" , expression , dataType , ")" ;dataType &＃61; "BYTE" | "SHORT" | "INT" | "LONG" | "FLOAT" | "DOUBLE" | "BOOLEAN" | "STRING" | "DECIMAL" | "SQL_DATE" | "SQL_TIME" | "SQL_TIMESTAMP" | "INTERVAL_MONTHS" | "INTERVAL_MILLIS" | ( "MAP" , "(" , dataType , "," , dataType , ")" ) | ( "PRIMITIVE_ARRAY" , "(" , dataType , ")" ) | ( "OBJECT_ARRAY" , "(" , dataType , ")" ) ;suffixAs &＃61; composite , ".as(" , fieldReference , ")" ;prefixAs &＃61; "as(" , expression, fieldReference , ")" ;suffixIf &＃61; composite , ".?(" , expression , "," , expression , ")" ;prefixIf &＃61; "?(" , expression , "," , expression , "," , expression , ")" ;suffixDistinct &＃61; composite , "distinct.()" ;prefixDistinct &＃61; functionIdentifier , ".distinct" , [ "(" , [ expression , { "," , expression } ] , ")" ] ;suffixFunctionCall &＃61; composite , "." , functionIdentifier , [ "(" , [ expression , { "," , expression } ] , ")" ] ;prefixFunctionCall &＃61; functionIdentifier , [ "(" , [ expression , { "," , expression } ] , ")" ] ;atom &＃61; ( "(" , expression , ")" ) | literal | fieldReference ;fieldReference &＃61; "*" | identifier ;nullLiteral &＃61; "nullOf(" , dataType , ")" ;timeIntervalUnit &＃61; "YEAR" | "YEAR_TO_MONTH" | "MONTH" | "QUARTER" | "WEEK" | "DAY" | "DAY_TO_HOUR" | "DAY_TO_MINUTE" | "DAY_TO_SECOND" | "HOUR" | "HOUR_TO_MINUTE" | "HOUR_TO_SECOND" | "MINUTE" | "MINUTE_TO_SECOND" | "SECOND" ;timePointUnit &＃61; "YEAR" | "MONTH" | "DAY" | "HOUR" | "MINUTE" | "SECOND" | "QUARTER" | "WEEK" | "MILLISECOND" | "MICROSECOND" ;over &＃61; composite , "over" , fieldReference ;overConstant &＃61; "current_row" | "current_range" | "unbounded_row" | "unbounded_row" ;timeIndicator &＃61; fieldReference , "." , ( "proctime" | "rowtime" ) ;


文字&＃xff1a;这里的文字是有效的Java文字。字符串文字可以使用单引号或双引号指定。复制引号以进行转义&＃xff08;例如“是我。”或“我”喜欢”狗。”&＃xff09;。

空文字&＃xff1a;空文字必须附加一个类型。使用nullOf&＃xff08;type&＃xff09;&＃xff08;例如nullOf&＃xff08;INT&＃xff09;&＃xff09;创建空值。

字段引用&＃xff1a;fieldReference指定数据中的一列&＃xff08;如果使用*&＃xff0c;则指定所有列&＃xff09;&＃xff0c;而functionIdentifier指定受支持的标量函数。列名和函数名遵循Java标识符语法。

函数调用&＃xff1a;指定为字符串的表达式也可以使用前缀表示法而不是后缀表示法来调用运算符和函数。

小数&＃xff1a;如果需要使用精确的数值或大的小数&＃xff0c;则Table API还支持Java的BigDecimal类型。在Scala Table API中&＃xff0c;小数可以由BigDecimal&＃xff08;“ 123456”&＃xff09;定义&＃xff0c;而在Java中&＃xff0c;可以通过附加“ p”来精确定义例如123456页

时间表示&＃xff1a;为了使用时间值&＃xff0c;Table API支持Java SQL的日期&＃xff0c;时间和时间戳类型。在Scala Table API中&＃xff0c;可以使用java.sql.Date.valueOf&＃xff08;“ 2016-06-27”&＃xff09;&＃xff0c;java.sql.Time.valueOf&＃xff08;“ 10:10:42”&＃xff09;或java.sql定义文字。Timestamp.valueOf&＃xff08;“ 2016-06-27 10&＃xff1a;10&＃xff1a;42.123”&＃xff09;。Java和Scala表API还支持调用“ 2016-06-27” .toDate&＃xff08;&＃xff09;&＃xff0c;“ 10:10:42” .toTime&＃xff08;&＃xff09;和“ 2016-06-27 10&＃xff1a;10&＃xff1a;42.123” .toTimestamp&＃xff08;&＃xff09;用于将字符串转换为时间类型。注意&＃xff1a;由于Java的时态SQL类型取决于时区&＃xff0c;因此请确保Flink Client和所有TaskManager使用相同的时区。

时间间隔&＃xff1a;时间间隔可以表示为月数&＃xff08;Types.INTERVAL_MONTHS&＃xff09;或毫秒数&＃xff08;Types.INTERVAL_MILLIS&＃xff09;。可以添加或减去相同类型的间隔&＃xff08;例如1.小时&＃43; 10分钟&＃xff09;。可以将毫秒间隔添加到时间点&＃xff08;例如“ 2016-08-10” .toDate &＃43; 5.days&＃xff09;。

Scala表达式:&＃xff1a;Scala表达式使用隐式转换。因此&＃xff0c;请确保将通配符导入org.apache.flink.table.api.scala._添加到程序中。如果文字不被视为表达式&＃xff0c;请使用.toExpr&＃xff08;如3.toExpr&＃xff09;强制转换文字。