请大家忽略该邮件,已找到问题所在 song wang <sdlcwangson...@gmail.com> 于2020年8月3日周一 上午10:16写道:
> 各位好, > > 我在使用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"); > > > >