光闪 SQL 单元测试：如何分配水印？

Question

我正在为使用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

Answer 1

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

1.使用< code>csv连接器，它与您的< code >值堆叠在一起，如下例所示。

2. 使用内置的数据生成连接器。由于您正在为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");

水印被声明为计算列，您可以选择使用多个水印策略。有关更多详细信息，请查看此文档页面。