版本: 1.10.1
引擎：flink
先创建的table表schema如下：
-- root
|-- address: STRING
|-- tum: TIMESTAMP(3)
|-- tum_count: BIGINT
table2表schema如下：
root
|-- name: STRING
|-- age: STRING
|-- field_time: TIMESTAMP(3)
|-- address: STRING
|-- proctimeField: TIMESTAMP(3)
其中proctimeField是PROCTIME
然后执行sql如下:
"insert into `table`\n" +
"select\n" +
" address,\n" +
" tumble_start(proctimeField, INTERVAL '10' SECOND) as tum,\n" +
" count(1) as tum_count \n" +
"from `table2\n" +
"group by \n" +
" tumble(proctimeField, INTERVAL '10' SECOND), address;
出现如下异常：
org.apache.flink.table.api.ValidationException: Field types of query result
and registered TableSink
`vstream-catalog`.`vstream-test`.`window_from_field_test_02` do not match.
Query result schema: [address: String, tum: TimeIndicatorTypeInfo(proctime),
tum_count: Long]
TableSink schema: [address: String, tum: LocalDateTime, tum_count: Long]
at
org.apache.flink.table.sinks.TableSinkUtils$.validateSink(TableSinkUtils.scala:64)
at
org.apache.flink.table.planner.StreamPlanner$$anonfun$2.apply(StreamPlanner.scala:146)
at
org.apache.flink.table.planner.StreamPlanner$$anonfun$2.apply(StreamPlanner.scala:145)
at scala.Option.map(Option.scala:146)
at
org.apache.flink.table.planner.StreamPlanner.org$apache$flink$table$planner$StreamPlanner$$translate(StreamPlanner.scala:145)
at
org.apache.flink.table.planner.StreamPlanner$$anonfun$translate$1.apply(StreamPlanner.scala:117)
at
org.apache.flink.table.planner.StreamPlanner$$anonfun$translate$1.apply(StreamPlanner.scala:117)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
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.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at
org.apache.flink.table.planner.StreamPlanner.translate(StreamPlanner.scala:117)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:697)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlUpdate(TableEnvironmentImpl.java:507)
at
org.apache.flink.vstream.sql.submit.SqlClientTest.compileSql(SqlClientTest.java:366)
at
org.apache.flink.vstream.sql.submit.SqlClientTest.test(SqlClientTest.java:208)
求大佬指点迷津
--
Sent from:
http://apache-flink.147419.n8.nabble.com/
Locked
