各位好, 我在使用flink 1.11.0 连接mysql时,只要查询 integer 类型字段就会报如下错误,查询其他bigint、varchar类型都正常,请问是哪里有问题呢?
报错信息: Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer at org.apache.flink.table.data.GenericRowData.getInt(GenericRowData.java:149) at org.apache.flink.table.data.RowData.get(RowData.java:257) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:156) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:123) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:50) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104) at org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:93) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201) 代码如下: StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); EnvironmentSettings settings = EnvironmentSettings.newInstance() .useBlinkPlanner() .inStreamingMode() .build(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings); tEnv.executeSql("create table base_price (\n" + " id INT,\n" + " mall_source INT,\n" + " sku_id STRING,\n" + " pricing_type INT,\n" + " pricing_method INT,\n" + " pricing_rule_id INT,\n" + " price BIGINT,\n" + " marking_price BIGINT\n" + ")\n" + "WITH (\n" + " 'connector' = 'jdbc',\n" + " 'url' = 'jdbc:mysql://127.0.0.1:3306/business',\n" + " 'table-name' = 'base_price',\n" + " 'username' = 'user',\n" + " 'password' = 'password',\n" + " 'lookup.cache.ttl' = '1s',\n" + " 'lookup.cache.max-rows' = '10'\n" + ")"); TableResult result = tEnv.sqlQuery("select id from base_price").execute(); result.print(); env.execute("sql demo");