问题描述
我正在为使用match_recognize的Flink sql语句编写单元测试。我正在像这样设置测试数据
Table data = tEnv.fromValues(DataTypes.ROW(
DataTypes.FIELD("event_time",DataTypes.TIMESTAMP(3)),DataTypes.FIELD("foobar",DataTypes.STRING()),....
),row(...),row(...)
);
我有两个问题,
FLINK版本:1.11
解决方法
您遇到了Table API的当前限制:无法结合forValues
方法来定义水印和行时间属性;您需要一个连接器。有几种解决方法:
1。。请使用与csv
堆叠在一起的VALUES
连接器,如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了解更多详情。