java - 如何定义具有行时间属性的 apache flink 表
问题描述
我有 json 行作为我的数据,我想用它创建一个表。
StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(streamExecutionEnvironment, fsSettings);
String allEventsTable = "allEventsTable";
fsTableEnv.connect(new Kafka()
.version("0.11")
.topic("events")
.property("bootstrap.servers", "localhost:9092")
.property("group.id", "dummyquery").startFromLatest())
.withSchema(new Schema()
.field("rule_id", Types.INT)
.field("sourceAddress", Types.STRING)
.field("deviceProduct", Types.STRING)
.field("destHost", Types.STRING)
.field("extra", Types.STRING)
.field("rowtime", Types.SQL_TIMESTAMP)
.rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))
)
.withFormat(new Json().failOnMissingField(false).deriveSchema())
.inAppendMode()
.registerTableSource(allEventsTable);
Table result = fsTableEnv.sqlQuery("select * from allEventsTable where sourceAddress='12345431'");
DataStream alert = fsTableEnv.toAppendStream(result, Row.class);
alert.print();
但是,在运行作业时出现错误
Exception in thread "main" org.apache.flink.table.api.ValidationException: Field 'rowtime' could not be resolved by the field mapping.
at org.apache.flink.table.sources.TableSourceValidation.resolveField(TableSourceValidation.java:245)
at org.apache.flink.table.sources.TableSourceValidation.lambda$validateTimestampExtractorArguments$6(TableSourceValidation.java:202)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
附言。我正在使用 flink 1.9
我放入kafka主题事件的json数据就像
{"rule_id":"", "rowtime":"2020-07-23 13:10:13","sourceAddress":"12345433","deviceProduct":"234r5t", "destHost":"876543", "extra":"dummy"}
解决方案
恐怕这是一个错误。我创建了https://issues.apache.org/jira/browse/FLINK-15801来跟踪它。
如果您更改行时间定义中的字段名称之一,您应该能够解决它。更改逻辑字段的名称:
.field("timeAttribute", Types.SQL_TIMESTAMP)
.rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))
或起源的物理场:
.field("rowtime", Types.SQL_TIMESTAMP)
.rowtime(new Rowtime().timestampsFromField("timestamp").watermarksPeriodicBounded(2000))
推荐阅读
- python - 为什么我在使用 python binance 库时会出现此错误?
- node.js - 每次我尝试制作 React 应用程序时,它都在寻找资金并且存在一些漏洞。然后卡住
- android - onChanged LiveData 未调用,RecyclerView 保持为空
- r - 如何使用 ggplot2 为散点图绘制多条趋势线?
- c# - C# WPF Websocket - 无法与 NGINX 建立握手
- javascript - 从间隔内更改 setInterval 时间
- reactjs - Rxjs observable 被订阅了两次
- python - 带有循环的“列表索引超出范围”
- android - SwipeRefreshLayout setOnRefreshLayout 空引用片段
- image - 通过切换平面和旋转相机角度的 ThreeJS 图像过渡