流处理概念

Flink的 Table API和SQL支持是用于批处理和流处理的统一API。这意味着 Table API和SQL查询具有相同的语义，无论它们的输入是有界批量输入还是无界流输入。因为关系代数和SQL最初是为批处理而设计的，所以关于无界流输入的关系查询不像有界批输入上的关系查询那样容易理解。

在此页面上，我们将解释Flink关于流数据的关系API的概念，实际限制和特定于流的配置参数。

数据流上的关系查询

SQL和关系代数的设计并未考虑流数据。因此，关系代数（和SQL）与流处理之间的概念差距很小。

尽管存在这些差异，但使用关系查询和SQL处理流并非不可能。高级关系数据库系统提供称为物化视图的函数。物化视图定义为SQL查询，就像常规虚拟视图一样。与虚拟视图相比，物化视图缓存查询的结果，使得在访问视图时不需要评估查询。缓存的一个常见挑战是阻止缓存提供过时的结果。实例化视图在修改其定义查询的基表时会过时。Eager View Maintenance是一种更新实体化视图并在其基表更新后立即更新实体化视图的技术。

如果我们考虑以下内容，急切的视图维护和流上的SQL查询之间的联系就变得很明显：

• 数据库表是一个结果流的INSERT，UPDATE和DELETEDML语句，通常被称为更新日志流。
• 物化视图定义为SQL查询。为了更新视图，查询将持续处理视图基本关系的更改日志流。
• 物化视图是流式SQL查询的结果。考虑到这些要点，我们将在下一节介绍Flink的动态表概念。

动态表和连续查询

动态表是Flink的 Table API和SQL支持流数据的核心概念。与表示批处理数据的静态表相比，动态表随时间而变化。可以像静态批处理表一样查询它们。查询动态表会产生连续查询。连续查询永远不会终止并生成动态表作为结果。查询不断更新其（动态）结果表以反映其输入（动态）表的更改。实质上，对动态表的连续查询与物化视图的定义查询非常相似。

值得注意的是，连续查询的结果始终在语义上等同于在输入表的SNAPSHOT上以批处理模式执行的相同查询的结果。

下图显示了流，动态表和连续查询的关系：

[   user:  VARCHAR,   // the name of the user  cTime: TIMESTAMP, // the time when the URL was accessed  url:   VARCHAR    // the URL that was accessed by the user]

SELECT user, COUNT(url)FROM clicksGROUP BY user;

SELECT user, RANK() OVER (ORDER BY lastLogin)FROM (  SELECT user, MAX(cTime) AS lastAction FROM clicks GROUP BY user);

• Upsert流： upsert流是一种包含两种消息，upsert消息和删除消息的流。转换为upsert流的动态表需要（可能是复合的）唯一键。具有唯一键的动态表通过编码转换为动态表，INSERT并UPDATE更改为upsert消息并DELETE更改为删除消息。流消耗 算子需要知道唯一键属性才能正确应用消息。与收回流的主要区别在于，UPDATE使用单个消息对更改进行编码，因此更有效。下图显示了动态表到upsert流的转换。

DataStream在Common Concepts页面上讨论了将动态表转换为a的API 。请注意，将动态表格转换为a时，仅支持附加和撤消流DataStream。TableSources和TableSinks页面TableSink讨论了向外部系统发出动态表的接口。

时间属性

Flink能够根据不同的时间概念处理流数据。

• 处理时间是指执行相应 算子操作的机器的系统时间（也称为“挂钟时间”）。
• 事件时间是指基于附加到每一行的时间戳来处理流数据。时间戳可以在事件发生时进行编码。
• 摄取时间是事件进入Flink的时间; 在内部，它与事件时间类似。有关Flink中时间处理的更多信息，请参阅有关事件时间和水印的简介。

表程序要求为流环境指定相应的时间特性：

• Java
• Scala

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // default
// alternatively:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) // default
// alternatively:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

Table API和SQL中基于时间的 算子操作（如窗口）需要有关时间概念及其来源的信息。因此，表可以提供逻辑时间属性，用于指示时间和访问表程序中的相应时间戳。

时间属性可以是每个表模式的一部分。它们是在从a创建表时定义的，DataStream或在使用时预定义的TableSource。一旦在开头定义了时间属性，它就可以作为字段引用，并可以用于基于时间的 算子操作。

只要时间属性未被修改并且只是从查询的一部分转发到另一部分，它仍然是有效的时间属性。时间属性的行为类似于常规时间戳，可以访问以进行计算。如果在计算中使用时间属性，则它将具体化并成为常规时间戳。常规时间戳不与Flink的时间和水印系统配合，因此不能再用于基于时间的 算子操作。

处理时间

处理时间允许表程序根据本地机器的时间产生结果。这是最简单的时间概念，但不提供决定论。它既不需要时间戳提取也不需要水印生成。

有两种方法可以定义处理时间属性。

在DataStream到表转换期间

处理时间属性是.proctime在架构定义期间使用属性定义的。time属性必须仅通过附加逻辑字段扩展物理模式。因此，它只能在模式定义的末尾定义。

• Java
• Scala

DataStream<Tuple2<String, String>> stream = ...;
// declare an additional logical field as a processing time attribute
Table table = tEnv.fromDataStream(stream, "Username, Data, UserActionTime.proctime");
WindowedTable windowedTable = table.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));

val stream: DataStream[(String, String)] = ...
// declare an additional logical field as a processing time attribute
val table = tEnv.fromDataStream(stream, 'UserActionTimestamp, 'Username, 'Data, 'UserActionTime.proctime)
val windowedTable = table.window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

使用TableSource

处理时间属性由TableSource实现DefinedProctimeAttribute接口的a定义。逻辑时间属性附加到由返回类型定义的物理模式TableSource。

• Java
• Scala

// define a table source with a processing attribute
public class UserActionSource implements StreamTableSource<Row>, DefinedProctimeAttribute {
    @Override
    public TypeInformation<Row> getReturnType() {
        String[] names = new String[] {"Username" , "Data"};
        TypeInformation[] types = new TypeInformation[] {Types.STRING(), Types.STRING()};
        return Types.ROW(names, types);
    }
    @Override
    public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
        // create stream 
        DataStream<Row> stream = ...;
        return stream;
    }
    @Override
    public String getProctimeAttribute() {
        // field with this name will be appended as a third field 
        return "UserActionTime";
    }
}
// register table source
tEnv.registerTableSource("UserActions", new UserActionSource());
WindowedTable windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));

// define a table source with a processing attribute
class UserActionSource extends StreamTableSource[Row] with DefinedProctimeAttribute {
    override def getReturnType = {
        val names = Array[String]("Username" , "Data")
        val types = Array[TypeInformation[_]](Types.STRING, Types.STRING)
        Types.ROW(names, types)
    }
    override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
        // create stream
        val stream = ...
        stream
    }
    override def getProctimeAttribute = {
        // field with this name will be appended as a third field
        "UserActionTime"
    }
}
// register table source
tEnv.registerTableSource("UserActions", new UserActionSource)
val windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

活动时间

事件时间允许表程序根据每个记录中包含的时间生成结果。即使在无序事件或延迟事件的情况下，这也允许一致的结果。当从持久存储中读取记录时，它还确保表程序的可重放结果。

此外，事件时间允许批处理和流式环境中的表程序的统一语法。流式传输环境中的时间属性可以是批处理环境中的记录的常规字段。

为了处理乱序事件并区分流处理中的准时和迟到事件，Flink需要从事件中提取时间戳并在时间上做出某种进展（所谓的水印）。

可以在DataStream-to-Table转换期间或使用TableSource定义事件时间属性。

在DataStream到表转换期间

.rowtime在架构定义期间使用属性定义事件时间属性。必须在转换后分配时间戳和水印DataStream。

将a转换DataStream为a 时，有两种定义时间属性的方法Table。根据指定的.rowtime字段名称是否存在于模式中DataStream，时间戳字段也是

• 作为新架构附加到架构或

• 替换现有字段。在任何一种情况下，事件时间时间戳字段都将保存事件时间时间戳的值DataStream。

• Java

• Scala

// Option 1:
// extract timestamp and assign watermarks based on knowledge of the stream
DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);
// declare an additional logical field as an event time attribute
Table table = tEnv.fromDataStream(stream, "Username, Data, UserActionTime.rowtime");
// Option 2:
// extract timestamp from first field, and assign watermarks based on knowledge of the stream
DataStream<Tuple3<Long, String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);
// the first field has been used for timestamp extraction, and is no longer necessary
// replace first field with a logical event time attribute
Table table = tEnv.fromDataStream(stream, "UserActionTime.rowtime, Username, Data");
// Usage:
WindowedTable windowedTable = table.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));

// Option 1:
// extract timestamp and assign watermarks based on knowledge of the stream
val stream: DataStream[(String, String)] = inputStream.assignTimestampsAndWatermarks(...)
// declare an additional logical field as an event time attribute
val table = tEnv.fromDataStream(stream, 'Username, 'Data, 'UserActionTime.rowtime)
// Option 2:
// extract timestamp from first field, and assign watermarks based on knowledge of the stream
val stream: DataStream[(Long, String, String)] = inputStream.assignTimestampsAndWatermarks(...)
// the first field has been used for timestamp extraction, and is no longer necessary
// replace first field with a logical event time attribute
val table = tEnv.fromDataStream(stream, 'UserActionTime.rowtime, 'Username, 'Data)
// Usage:
val windowedTable = table.window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

使用TableSource

事件时间属性由TableSource实现DefinedRowtimeAttribute接口的a定义。该getRowtimeAttribute()方法返回一个现有字段的名称，该字段包含表的事件时间属性，并且是类型LONG或TIMESTAMP。

此外，方法DataStream返回的getDataStream()必须分配与定义的时间属性对齐的水印。请注意，忽略（DataStream由a分配的TimestampAssigner）时间戳。只有TableSource's rowtime属性的值是相关的。

• Java
• Scala

// define a table source with a rowtime attribute
public class UserActionSource implements StreamTableSource<Row>, DefinedRowtimeAttribute {
    @Override
    public TypeInformation<Row> getReturnType() {
        String[] names = new String[] {"Username", "Data", "UserActionTime"};
        TypeInformation[] types =
            new TypeInformation[] {Types.STRING(), Types.STRING(), Types.LONG()};
        return Types.ROW(names, types);
    }
    @Override
    public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
        // create stream 
        // ...
        // assign watermarks based on the "UserActionTime" attribute
        DataStream<Row> stream = inputStream.assignTimestampsAndWatermarks(...);
        return stream;
    }
    @Override
    public String getRowtimeAttribute() {
        // Mark the "UserActionTime" attribute as event-time attribute.
        return "UserActionTime";
    }
}
// register the table source
tEnv.registerTableSource("UserActions", new UserActionSource());
WindowedTable windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));

// define a table source with a rowtime attribute
class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttribute {
    override def getReturnType = {
        val names = Array[String]("Username" , "Data", "UserActionTime")
        val types = Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.LONG)
        Types.ROW(names, types)
    }
    override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
        // create stream
        // ...
        // assign watermarks based on the "UserActionTime" attribute
        val stream = inputStream.assignTimestampsAndWatermarks(...)
        stream
    }
    override def getRowtimeAttribute = {
        // Mark the "UserActionTime" attribute as event-time attribute.
        "UserActionTime"
    }
}
// register the table source
tEnv.registerTableSource("UserActions", new UserActionSource)
val windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

查询配置

Table API和SQL查询具有相同的语义，无论它们的输入是有界批量输入还是无界流输入。在许多情况下，对流输入的连续查询能够计算与离线计算结果相同的准确结果。然而，这在一般情况下是不可能的，因为连续查询必须限制它们维护的状态的大小，以避免耗尽存储并且能够在很长一段时间内处理无界流数据。因此，连续查询可能只能提供近似结果，具体取决于输入数据和查询本身的特征。

Flink的 Table API和SQL接口提供参数来调整连续查询的准确性和资源消耗。参数通过QueryConfig对象指定。在QueryConfig可以从获得TableEnvironment和被传递回一个时Table被转换，即，当它被变换成数据流或经由TableSink发射。

• Java
• Scala

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
// obtain query configuration from TableEnvironment
StreamQueryConfig qConfig = tableEnv.queryConfig();
// set query parameters
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24));
// define query
Table result = ...
// create TableSink
TableSink<Row> sink = ...
// emit result Table via a TableSink
result.writeToSink(sink, qConfig);
// convert result Table into a DataStream<Row>
DataStream<Row> stream = tableEnv.toAppendStream(result, Row.class, qConfig);

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)
// obtain query configuration from TableEnvironment
val qConfig: StreamQueryConfig = tableEnv.queryConfig
// set query parameters
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))
// define query
val result: Table = ???
// create TableSink
val sink: TableSink[Row] = ???
// emit result Table via a TableSink
result.writeToSink(sink, qConfig)
// convert result Table into a DataStream[Row]
val stream: DataStream[Row] = result.toAppendStream[Row](qConfig)

在下文中，我们将描述它们的参数QueryConfig以及它们如何影响查询的准确性和资源消耗。

空闲状态保存时间

许多查询在一个或多个键属性上聚合或连接记录。当在流上执行此类查询时，连续查询需要收集记录或维护每个键的部分结果。如果输入流的关键域正在发展，即活动键值随时间变化，则随着观察到越来越多的不同键，连续查询累积越来越多的状态。但是，经常在一段时间后Keys变为非活动状态，并且它们的相应状态变得陈旧且无用。

例如，以下查询计算每个会话的单击次数。

SELECT sessionId, COUNT(*) FROM clicks GROUP BY sessionId;

该sessionId属性用作分组键，连续查询维护sessionId其观察的每个键的计数。该sessionId属性随着时间的推移而发展，并且sessionId值仅在会话结束之前有效，即，在有限的时间段内。但是，连续查询无法知道此属性，sessionId并期望每个sessionId值都可以在任何时间点发生。它维护每个观察sessionId值的计数。因此，随着sessionId观察到越来越多的值，查询的总状态大小不断增长。

在空闲状态保持时间参数定义多久一个键的状态得以保持，而它被删除之前被更新。对于前面的示例查询，sessionId只要在配置的时间段内没有更新，就会删除a的计数。

通过删除键的状态，连续查询完全忘记它之前已经看过这个键。如果处理了具有其状态已被删除的Keys的记录，则该记录将被视为具有相应Keys的第一个记录。对于上面的例子，这意味着a的计数sessionId将再次开始0。

配置空闲状态保存时间有两个参数：

• 最小空闲状态保持时间定义多久它被删除之前非活动键的状态至少保持。

• 在最大空闲状态保持时间定义多久它被删除前的非激活键的状态最多保持。参数指定如下：

• Java

• Scala

StreamQueryConfig qConfig = ...
// set idle state retention time: min = 12 hours, max = 24 hours
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24));

val qConfig: StreamQueryConfig = ???
// set idle state retention time: min = 12 hours, max = 24 hours
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))

清理状态需要额外副本记录，而且记录使得minTime和maxTime不是那么浪费时间和空间。minTime和之间的差异maxTime必须至少为5分钟。