flink sql read hbase sink mysql data type not matc
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Field types of query result and registered TableSink default_catalog.default_database.user_age do not match. Query schema: [rowkey: STRING, cf: ROW<age INT>] Sink schema: [rowkey: STRING, age: INT] at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138) at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:664) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:895) at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:968) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:968) Caused by: org.apache.flink.table.api.ValidationException: Field types of query result and registered TableSink default_catalog.default_database.user_age do not match.
query:
val users = new HBaseTableSource(hConf, "user_hbase5") users.setRowKey("rowkey", classOf[String]) // currency as the primary key users.addColumn("cf", "age", classOf[Integer])
streamTableEnv.registerTableSource("users", users)
streamTableEnv.sqlUpdate( """ | |CREATE TABLE user_age ( | rowkey VARCHAR, | age INT |) WITH ( | 'connector.type' = 'jdbc', | 'connector.write.flush.max-rows' = '1' |) |""".stripMargin)
streamTableEnv.sqlUpdate( """ | |insert into user_age |SELECT * |FROM | users | |""".stripMargin)
*来自志愿者整理的flink邮件归档
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
这个错误信息表明在Apache Flink程序中，查询结果的数据结构与已注册的TableSink（即输出表user_age）的数据结构不匹配。具体来说，问题出在字段类型上。
你的查询结果和sink定义有以下差异：
- 查询结果的模式是：
[rowkey: STRING, cf: ROW<age INT>] - 而你注册的TableSink（JDBC sink）的模式是：
[rowkey: STRING, age: INT]
问题在于查询结果中的cf是一个ROW类型，包含一个名为age的INT字段，而sink期望直接接收一个名为age的INT字段，没有嵌套的ROW结构。
为了解决这个问题，你需要调整查询或者TableSink的定义，以确保两边的模式相匹配。一种解决方法是在查询时直接提取cf.age字段，而不是使用*选择所有列。修改插入语句如下：
streamTableEnv.sqlUpdate(
"""
INSERT INTO user_age
SELECT rowkey, cf.age AS age
FROM users
""".stripMargin)
这样修改后，查询结果将只包含rowkey和直接的age字段，与user_age表的定义相匹配，应该可以解决类型不匹配的问题。
