Flink SQL 单元测试：如何分配水印？答案

【问题标题】：Flink SQL Unit Testing: How to Assign Watermark?Flink SQL 单元测试：如何分配水印？
【发布时间】：2020-09-28 16:34:43
【问题描述】：

我正在为使用 match_recognize 的 Flink SQL 语句编写单元测试。我正在设置这样的测试数据

Table data = tEnv.fromValues(DataTypes.ROW(
  DataTypes.FIELD("event_time", DataTypes.TIMESTAMP(3)),
  DataTypes.FIELD("foobar", DataTypes.STRING()),
  ....
  ),
  row(...),
  row(...)
);

我有两个问题，

如何将 event_time 指定为水印字段？（表示行时间）
不太重要，给创建的表起一个有意义的名称？

FLINK 版本：1.11

【问题讨论】：

标签： apache-flink flink-streaming flink-sql flink-table-api

【解决方案1】：

您遇到了 Table API 的当前限制：无法结合 forValues 方法定义水印和行时间属性；你需要一个连接器。有几个选项可以解决它：

1. 使用与VALUES 叠加的csv 连接器，如this example 中所示。

2. 使用内置的DataGen connector。由于您正在为 CEP 进行单元测试，我想您希望对生成的数据进行一定程度的控制，因此这可能不是一个可行的选择。无论如何，我想我会提到它。

注意： 推荐使用 SQL DDL 语法从 Flink 1.10 创建表。这将使您尝试做的两件事（即定义水印和命名您的表格）更加简单：

tEnv.executeSql("CREATE TABLE table_name (\n" +
                "             event_time TIMESTAMP(3),\n" +
                "             foobar STRING \n" +
                "             WATERMARK FOR event_time AS event_time\n" +
                ") WITH (...)"
);

Table data = tEnv.from("table_name");

水印被声明为计算列，您可以选择使用多种水印策略。详情请查看this documentation page。

【讨论】：

我明白了。感谢您的回复。