java - 在数据集中转换 Flink CEP 表时出错目的
问题描述
我创建了一个BatchTableEnviroment
inApache Flink
并创建了一个Table
加载数据的对象。现在我想搜索一些模式。我正在使用Detecting patterns in Tables CEP library
. 下面查询的任务是找出 amGroup
中avgResult
没有低于某个阈值的最长周期。其中 amGroup
是Integer
100、200、300 等值,Avgresult
是双精度值。当我编译查询部分时,我没有收到任何错误。将 转换为 时出现Table
错误DataSet<row>
。在查询下方,您可以看到错误消息。
ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = BatchTableEnvironment.create(fbEnv);
Table trendTable = tableEnv.sqlQuery(
" SELECT * " +
" FROM tableAvg " +
" MATCH_RECOGNIZE(" +
" PARTITION BY mType " +
" ORDER BY mGroup " +
" MEASURES " +
" FIRST(A.mGroup) as startGr, " +
" LAST(A.mGroup) as endGr, " +
" A.avgResult as avgRes" +
" ONE ROW PER MATCH " +
" AFTER MATCH SKIP PAST LAST ROW " +
" PATTERN (A+ B) " +
" DEFINE " +
" A AS A.avgResult < 50 " +
") "
);
tableEnv.registerTable("TrendTable", trendTable);
DataSet<Row> result = tableEnv.toDataSet(trendTable, Row.class);
/////////////////////ERROR MESSAGE BELOW
Exception in thread "main" org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query:
FlinkLogicalMatch(partition=[[$1]], order=[[2]], outputFields=[[mType, startGr, endGr, avgRes]], allRows=[false], after=[FLAG(SKIP PAST LAST ROW)], pattern=[(PATTERN_QUANTIFIER(_UTF-16LE'A', 1, -1, false), _UTF-16LE'B')], isStrictStarts=[false], isStrictEnds=[false], subsets=[[]], patternDefinitions=[[<(PREV(A.$0, 0), 50)]], inputFields=[[sumResult, mType, EXPR$2]])
FlinkLogicalSort(sort0=[$2], dir0=[ASC])
FlinkLogicalCalc(expr#0..5=[{inputs}], expr#6=[/($t2, $t3)], expr#7=[99], expr#8=[>($t5, $t7)], sumResult=[$t6], mType=[$t1], EXPR$2=[$t4], $condition=[$t8])
FlinkLogicalAggregate(group=[{0, 1}], agg#0=[SUM($2)], agg#1=[SUM($3)], agg#2=[MAX($4)], agg#3=[COUNT()])
FlinkLogicalCalc(expr#0..2=[{inputs}], expr#3=[1], expr#4=[-($t0, $t3)], expr#5=[100], expr#6=[/($t4, $t5)], expr#7=[1.0:DECIMAL(2, 1)], $f0=[$t6], mType=[$t1], mValue=[$t2], $f3=[$t7], mID=[$t0])
FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, H]], fields=[mID, dateTime, mValue, unixDateTime, mType], source=[CsvTableSource(read fields: mID, mType, mValue)])
This exception indicates that the query uses an unsupported SQL feature.
Please check the documentation for the set of currently supported SQL features.
at org.apache.flink.table.plan.Optimizer.runVolcanoPlanner(Optimizer.scala:245)
at org.apache.flink.table.plan.Optimizer.optimizePhysicalPlan(Optimizer.scala:170)
at org.apache.flink.table.plan.BatchOptimizer.optimize(BatchOptimizer.scala:57)
at org.apache.flink.table.api.internal.BatchTableEnvImpl.translate(BatchTableEnvImpl.scala:280)
at org.apache.flink.table.api.java.internal.BatchTableEnvironmentImpl.toDataSet(BatchTableEnvironmentImpl.scala:71)
at StreamTableEnv.main(StreamTableEnv.java:169)
解决方案
CEP 库和 MATCH_RECOGNIZE 仅在流 API 之上工作(而不是批处理),这意味着您需要使用 aStreamTableEnvironment
而不是 a BatchTableEnviroment
。
推荐阅读
- ios - 如何使用 Swift 从 iOS 中的 mimetype 获取文件扩展名
- java - Android:如何获取可绘制的 xml 名称
- plsql - PLSQL 触发器编译错误
- mysql - 如何将 HTML5 输出标签数据添加到 MYSQL?
- javascript - 为什么 mysql 在 node.js 中的多语句选项增加了 SQL 注入的范围?
- php - 如何创建在线论坛?
- npm - 如何更新本地模块?
- javascript - 将数组中的某些字符串大写
- hadoop - 本地系统的文件何时移动到 HDFS
- text-to-speech - 是否可以在使用 SpeechSynthesisUtterance API 时选择正在阅读的单词?