我建了个 issue 跟进这个功能：https://issues.apache.org/jira/browse/FLINK-20281
On Mon, 23 Nov 2020 at 10:35, Jark Wu <[email protected]> wrote: > Flink SQL 的 window agg 目前不支持输入含有更新和删除消息。 > 你可以使用非 window 聚合来代替。 > > Btw，你可能说一下你的需求场景么？ 为什么需要在 CDC 上做 window 操作呢？ > > Best, > Jark > > On Mon, 23 Nov 2020 at 10:28, jy l <[email protected]> wrote: > >> Hi： >> 我使用CDC组件debezium将MySQL中的维表数据实时发送到kafka，然后使用flinkSQL消费kafka的数据，具体流程如下： >> [image: image.png] >> [image: image.png] >> 分组计算的SQL如下： >> [image: image.png] >> 在执行计算时，报了如下异常： >> Exception in thread "main" org.apache.flink.table.api.TableException: >> GroupWindowAggregate doesn't support consuming update and delete changes >> which is produced by node TableSourceScan(table=[[default_catalog, >> default_database, t_order, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/($1, >> 1000))), 3000:INTERVAL SECOND)]]], fields=[id, timestamps, >> orderInformationId, userId, categoryId, productId, price, productCount, >> priceSum, shipAddress, receiverAddress]) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.createNewNode(FlinkChangelogModeInferenceProgram.scala:380) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.visit(FlinkChangelogModeInferenceProgram.scala:298) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.visitChild(FlinkChangelogModeInferenceProgram.scala:337) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.$anonfun$visitChildren$1(FlinkChangelogModeInferenceProgram.scala:326) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.$anonfun$visitChildren$1$adapted(FlinkChangelogModeInferenceProgram.scala:325) >> at >> scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) >> at scala.collection.immutable.Range.foreach(Range.scala:155) >> at scala.collection.TraversableLike.map(TraversableLike.scala:233) >> at scala.collection.TraversableLike.map$(TraversableLike.scala:226) >> at scala.collection.AbstractTraversable.map(Traversable.scala:104) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.visitChildren(FlinkChangelogModeInferenceProgram.scala:325) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.visit(FlinkChangelogModeInferenceProgram.scala:275) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.visitChild(FlinkChangelogModeInferenceProgram.scala:337) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.$anonfun$visitChildren$1(FlinkChangelogModeInferenceProgram.scala:326) >> at >> org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyModifyKindSetTraitVisitor.$anonfun$visitChildren$1$adapted(FlinkChangelogModeInferenceProgram.scala:325) >> at >> scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) >> >> 我的初步理解是FlinkSQL 不支持这样CDC这样由插入、更新、删除的流的分组聚合。 >> 那面对我这样的情况，该用什么方案来解决？ >> 望知道的各位告知一下，感谢！ >> >> 祝好 >> >>
