Field types of query result and registered TableSink 'table' do not match.

Sun, 27 Sep 2020 22:33:43 -0700 

版本: 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/

回复