Flink表异常:只能在时间属性列上定义窗口聚集，但遇到时间戳(6)

Question

我正在使用flink 1.12.0。尝试将数据流转换为表A并在tableA上运行sql查询以在如下窗口上聚合。我使用f2列作为其时间戳数据类型字段。

    EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, fsSettings);

    Properties props = new Properties();

    props.setProperty("bootstrap.servers", BOOTSTRAP_SERVER);
    props.setProperty("schema.registry.url", xxx);
    props.setProperty("group.id", "test");
    props.setProperty("value.deserializer", KafkaAvroDeserializer.class.getName());

    props.put("client.id", "flink-kafka-example");

    FlinkKafkaConsumer<PlaybackListening> kafkaConsumer = new FlinkKafkaConsumer<>(
            "test-topic",
            ConfluentRegistryAvroDeserializationSchema.forSpecific(
                    Avrotest.class, prodSchemaRegistryURL),
            props);

    DataStreamSource<Avrotest> stream =
            env.addSource(kafkaConsumer);
    Table tableA = tEnv.fromDataStream(stream, $("f0"), $("f1"),$("f2"));
    Table result =
            tEnv.sqlQuery("SELECT f0, sum(f1),f2 FROM "
                    + tableA + " GROUP BY TUMBLE(f2, INTERVAL '1' HOUR) ,f1" );

    
    tEnv.toAppendStream(result,user.class).print();

    env.execute("Flink kafka test");
}

当我执行上述代码时，我得到

线程“main”org.apache.flink.table.api中出现异常。TableException:只能在时间属性列上定义窗口聚合，但遇到时间戳（6）。在org.apache.flink.table.planner.plan.rules.logical.StreamLogicalWindowAggregateGroup表达式（StreamLogicalWindowsAggregaterule.scala:50）和org.apach.flink.table.plan.rules.LogicalWindowaggregateBase.onMatch（LogicalWindowAGgregaterBase.scala:81）中，在org.apache.calcite.plan.AbstractRelopPlanner.fireRule（AbstractReplopLanner.java:333）中org.apache.calcite.plan.hep.HepPlanner.applyRules（hepplanter.java:542）位于org.apach.calciite.plan.hep.heppanner.applyroles（HepPlanner.java:407），org.apaache.calcites.plan.hep.HepPlanner.executeInstruction（HepPlanner.java:243），org.apache.calcite.plan.HepInstruction$RuleInstance.execute（hepInstitution.java:127）

Answer 1

3个步骤：

1. 首次分配分配时间戳和水印

你有几种策略。

例如：

 WatermarkStrategy<Row> customTime = WatermarkStrategy
                .<Row>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((event, timestamp) -> (long) event.getField("f2"));

env.addSource().assignTimestampsAndWatermarks(customTime)

Table tableA = tEnv.fromDataStream(stream, $("f0"), $("f1"),$("f2").rowtime());

Answer 2

若要使用表 API 在数据流上执行事件时间窗口化，需要先分配时间戳和水印。您应该在从DataStream调用之前执行此操作。

使用Kafka，通常最好直接在FlinkKafkaConsumer上调用depodTimestampsAndWatermarks。有关详细信息，请参阅水印文档、kafka 连接器文档和 Flink SQL 文档。