flink 1.11.0 conenctor-jdbc 报错

[wangsong2]

各位好，

我在使用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");

[wangsong2]

请大家忽略该邮件，已找到问题所在

song wang <[hidden email]> 于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");
>
>
>
>

[Leonard Xu]

Hi

> 在 2020年8月3日，10:16，song wang <[hidden email]> 写道：
>
> 查询 integer

如果MySQL中数据类型是 INT UNSIGNED，Flink 中 对应的类型是 BIGINT,
你检查下是不是这个原因，类型映射可以参考[1]

Best
Leonard
[1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#data-type-mapping <https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#data-type-mapping>

[wangsong2]

对，就是这个原因

Leonard Xu <[hidden email]> 于2020年8月3日周一 上午10:26写道：

> Hi
>
> > 在 2020年8月3日，10:16，song wang <[hidden email]> 写道：
> >
> > 查询 integer
>
> 如果MySQL中数据类型是 INT UNSIGNED，Flink 中 对应的类型是 BIGINT,
> 你检查下是不是这个原因，类型映射可以参考[1]
>
> Best
> Leonard
> [1]
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#data-type-mapping
> <
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#data-type-mapping
> >