在数据处理中时间处理是一个重要的任务。但是与此同时，处理不同的时区、日期和时间是一个日益复杂的任务。

在 Flink 1.13 中，我们投入了大量的精力来简化时间函数的使用。我们调整了时间相关函数的返回类型使其更加精确，例如 PROCTIME()，CURRENT_TIMESTAMP() 和 NOW()。

其次，用户现在还可以基于一个 TIMESTAMP_LTZ 类型的列来定义 Event Time 属性，从而可以优雅的在窗口处理中支持夏令时。

用户可以参考 Release Note 来查看该部分的完整变更。

PyFlink 核心优化

这个版本对 PyFlink 的改进主要是使基于 Python 的 DataStream API 与 Table API 与 Java/scala 版本的对应功能更加一致。

Python DataStream API 中的有状态算子

在 Flink 1.13 中，Python 程序员可以享受到 Flink 状态处理 API 的所有能力。在 Flink 1.12 版本重构过的 Python DataStream API 现在已经拥有完整的状态访问能力，从而使用户可以将数据的信息记录到 state 中并且在后续访问。

带状态的处理能力是许多依赖跨记录状态共享（例如 Window Operator）的复杂数据处理场景的基础。

以下例子展示了一个自定义的计算窗口的实现：

class CountWindowAverage(FlatMapFunction):

def __init__(self, window_size):

self.window_size = window_size

def open(self, runtime_context: RuntimeContext):

descriptor = ValueStateDescriptor("average", Types.TUPLE([Types.LONG(), Types.LONG()]))

self.sum = runtime_context.get_state(descriptor)

def flat_map(self, value):

current_sum = self.sum.value()

if current_sum is None:

current_sum = (0, 0)

# update the count

current_sum = (current_sum[0] + 1, current_sum[1] + value[1])

# if the count reaches window_size, emit the average and clear the state

if current_sum[0] >= self.window_size:

self.sum.clear()

yield value[0], current_sum[1] // current_sum[0]

else:

self.sum.update(current_sum)

ds = ... # type: DataStream

ds.key_by(lambda row: row[0]) \

.flat_map(CountWindowAverage(5))

PyFlink DataStream API 中的用户自定义窗口

Flink 1.13 中 PyFlink DataStream 接口增加了对用户自定义窗口的支持，现在用户可以使用标准窗口之外的窗口定义。

由于窗口是处理无限数据流的核心机制 （通过将流切分为多个有限的『桶』），这一功能极大的提高的 API 的表达能力。

PyFlink Table API 中基于行的操作

Python Table API 现在支持基于行的操作，例如用户对行数据的自定义函数。这一功能使得用户可以使用非内置的数据处理函数。

一个使用 map() 操作的 Python Table API 示例如下：

@udf(result_type=DataTypes.ROW(

[DataTypes.FIELD("c1", DataTypes.BIGINT()),