Последующая оконная группировка в Flink TableAPI приводит к RuntimeException
Я работаю над проектом по группировке / суммированию потоков графов с использованием API таблиц Apache Flink (1.12.0). В нашем алгоритме мы сначала обрабатываем вершины, т. Е. Группируем их и объединяем некоторые свойства. Вот фрагмент моей заявки:
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
final StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, bsSettings);
Timestamp t1 = new Timestamp(1619511681000L);
Timestamp t2 = ...
// Input data, StreamVertex(vertex_id, vertex_label, vertex_properties, event_time)
StreamVertex v11 = new StreamVertex("v1", "A", Properties.create(), t1);
StreamVertex v12 = ...
// Put input data to table
Table vertices = streamTableEnvironment.fromDataStream(
env.fromElements(v11, v12, v21, v22, v31, v32, v41, v42).assignTimestampsAndWatermarks(WatermarkStrategy
.<StreamVertex>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp) -> event.getEventTime().getTime())),
// Expressions with declaration of 'event_time' as rowtime
$(ID), $(LABEL), $(PROPERTIES), $(EVENT_TIME).rowtime());
// 1. Prepare distinct vertices
Table preparedVertices = vertices
.window(Tumble.over(lit(10).seconds()).on($(EVENT_TIME)).as("w1"))
.groupBy($(ID), $(LABEL), $("w1"))
.select($(ID), $(LABEL), $("w1").rowtime().as("w1_rowtime"));
// preparedVertices.execute().print(); --> would work well
// 2. Group vertices by label and/or property values
Table groupedVertices = preparedVertices
.window(Tumble.over(lit(10).seconds()).on($("w1_rowtime")).as("w2"))
.groupBy($(LABEL), $("w2"))
.select(
call("CreateSuperElementId", $(LABEL)).as("super_id"),
$(LABEL).as("super_label"),
lit(1).count().as("super_count"),
$("w2").rowtime().as("w2_rowtime"));
// groupedVertices.execute().print(); --> would work well
groupedVertices
.select($("super_id"), $("w2_rowtime"))
.execute().print(); // --> throws exception
Первая оконная группировка предназначена для удаления дубликатов, поскольку несколько ребер могут иметь одинаковые исходные / целевые вершины (то есть одинаковый идентификатор, метку и т. Д.) С разными отметками времени (отметки времени выбираются с ребер). Вторая группировка используется для группировки отдельных вершин по заданной конфигурации и агрегирования ее значений.
Проблема: когда я использую
groupedVerticesна последующих этапах, например, при простом проецировании (см. последние строки моего фрагмента), я получаю следующее исключение. Похоже, что псевдоним
w1_rowtime больше не "регистрируется".
java.lang.RuntimeException: Error while applying rule StreamExecGroupWindowAggregateRule(in:LOGICAL,out:STREAM_PHYSICAL), args [rel#197:FlinkLogicalWindowAggregate.LOGICAL.any.None: 0.[NONE].[NONE](input=RelSubset#196,group={1},window=TumblingGroupWindow('w2, w1_rowtime, 10000),properties=EXPR$1)]
at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
at org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
at org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
at org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:64)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:62)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58)
at scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
at scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:57)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:79)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:286)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:165)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1267)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:703)
at org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:570)
at edu.leipzig.impl.algorithm.GraphStreamGroupingTest.testDoubleGrouping(GraphStreamGroupingTest.java:225)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:221)
at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.IllegalArgumentException: field [w1_rowtime] not found; input fields are: [vertex_id, vertex_label, EXPR$0]
at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:415)
at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:398)
at org.apache.flink.table.planner.plan.utils.AggregateUtil$.timeFieldIndex(AggregateUtil.scala:888)
at org.apache.flink.table.planner.plan.rules.physical.stream.StreamExecGroupWindowAggregateRule.convert(StreamExecGroupWindowAggregateRule.scala:81)
at org.apache.calcite.rel.convert.ConverterRule.onMatch(ConverterRule.java:167)
at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:229)
... 48 more