Apache flink 1.52行时时间戳为空

Question

我正在使用以下代码进行一些查询：

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    DataStream<Row> ds = SourceHelp.builder().env(env).consumer010(MyKafka.builder().build().kafkaWithWaterMark2())
            .rowTypeInfo(MyRowType.builder().build().typeInfo())
            .build().source4();
    //,proctime.proctime,rowtime.rowtime
    String sql1 = "select a,b,max(rowtime)as rowtime from user_device group by a,b";
    DataStream<Row> ds2 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device").fields("a,b,rowtime.rowtime")
            .rowTypeInfo(MyRowType.builder().build().typeInfo13())
            .sql(sql1).in(ds).build().result();

    ds2.print();
    // String sql2 = "select a,count(b) as b from user_device2 group by a";
    String sql2 = "select a,count(b) as b,HOP_END(rowtime,INTERVAL '5' SECOND,INTERVAL '30' SECOND) as c from user_device2 group by HOP(rowtime, INTERVAL '5' SECOND, INTERVAL '30' SECOND),a";
    DataStream<Row> ds3 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device2").fields("a,b,rowtime.rowtime")
            .rowTypeInfo(MyRowType.builder().build().typeInfo14())
            .sql(sql2).in(ds2).build().result();

    ds3.print();
    env.execute("test");

注意:对于sql1，我使用带有rowtime的max函数，它不起作用，并引发以下异常:

线程"main"org.apache.flink.runtime.client.JobExecutionException：java.lang.RuntimeException中的异常：Rowtime时间戳为空。请确保定义了正确的TimestampAssigner并且流环境使用EventTime时间特性。org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking（MiniCluster.java:625）org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute（LocalStreamEnvironment.java:123）com.aicaigroup.water.WaterTest.testRowtimeWithMoreSqls5（WaterTest.java:158）com.aicaigroup.water.WaterTest.main（WaterTest.java:20）引起的：java.lang.RuntimeException： Rowtime时间戳为空。请确保定义了正确的TimestampAssigner，并且流环境使用EventTime时间特性。在DataStreamSourceConversion$org.apache.flink.streaming.api.operators.AbstractStreamOperator（未知来源）在CountingOutput.collect（CRowOutputProcessRunner.scala:67）在org.apache.flink.streaming.api.operators.ProcessOperator.processElement（ProcessOperator.java:66）在org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator（OperatorChain.java:558）在org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect（OperatorChain.java:533）在org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect（OperatorChain.java:513）在org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect（OperatorChain.java:628）在org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect（OperatorChain.java:581）在24.processElement$org.apache.flink.table.runtime.CRowOutputProcessRunner.processElement（AbstractStreamoperator. java： 679）在org. apache. flink. stream. api. operator. AbstractStreamoperator$Counting Output. Collection（AbstractStreamoperator. java： 657）在org. apache. flink. stream. api. operator. TimestatamCollector. Collection（TimestampeColltor. java： 51）在com. aicaigroup. TableHelp1 Dollar. Process Element（TableHelp1 Dollar. java： 42）在com. aicaigroup. TableHelp1 Element（TableHelp. java）533）在org. apache. flink. stream. runtime. tasks.0014 atorChain$CopyingChainingOutput.集合（操作链. java： 513）在org. apache. flink. stream. api. operator. AbstractStream运算符$Counting Output.集合（AbstractStream运算符. java： 679）在org. apache. flink. stream. api.操作员. AbstractStream运算符$Counting Output.集合（AbstractStream运算符. java： 657）在org. apache. flink. stream. api. operator. TimestamedCollector.收集（Timestampe Collector. java： 51）在org. apache. flink. table. runtime.聚合. GroupAggProcessFunc. Element（GroupAggessFunc. scala： 151）

然后我试着这样更新sql1“从user_device中选择a，b，rowtime”，成功了。那么如何修复错误呢？第一个sql应该使用group by，第二个sql应该使用rowtime by timeWindow。3q

Answer 1

我从1.6开始眨眼，遇到像你这样的类似问题。通过这些步骤解决：

• 使用赋值时间戳和水印，只需使用默认和正常的实现BoundedOutOfOrdernessTimestampExtractor。您需要编写extTimestamp函数来提取时间戳值并在构造函数中声明窗口间隔。
• 在字段末尾追加proctime.proctimerowtime.rowtime（我正在使用fromDataStream（Flink 1.6）将stream转换为表）
• 如果您想使用存在字段作为行时间。例如，数据源字段是“a， clicktime， c”，您可以声明“a，clicktime.rowtime， c”

希望能帮到你。