You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by kcz <57...@qq.com.INVALID> on 2021/06/19 15:16:12 UTC
flink-1.13.1 sql error
sql如下：
CREATE TABLE user_behavior (
user_id BIGINT,
item_id BIGINT,
category_id BIGINT,
behavior STRING,
ts STRING
) WITH (
'connector' = 'kafka',
'topic' = 'user_behavior',
'scan.startup.mode' = 'latest-offset',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
);
select * from user_behavior;
pom.xml如下：
flink.version=1.13.1
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_2.11</artifactId>
<version>${flink.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.56</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.12</version>
</dependency>
报错log如下：
Caused by: java.io.IOException: Failed to deserialize JSON '"{\n \"app_time\": \"2021-06-14 10:00:00\",\n \"category_id\": 1,\n \"item_id\": 1,\n \"user_id\": 1,\n \"behavior\": \"pv\"\n}"'.
at org.apache.flink.formats.json.JsonRowDataDeserializationSchema.deserialize(JsonRowDataDeserializationSchema.java:112)
at org.apache.flink.formats.json.JsonRowDataDeserializationSchema.deserialize(JsonRowDataDeserializationSchema.java:50)
at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82)
at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113)
at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:179)
at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:142)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:826)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)
Caused by: java.lang.ClassCastException: org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode cannot be cast to org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode
at org.apache.flink.formats.json.JsonToRowDataConverters.lambda$createRowConverter$ef66fe9a$1(JsonToRowDataConverters.java:344)
at org.apache.flink.formats.json.JsonToRowDataConverters.lambda$wrapIntoNullableConverter$de0b9253$1(JsonToRowDataConverters.java:376)
at org.apache.flink.formats.json.JsonRowDataDeserializationSchema.convertToRowData(JsonRowDataDeserializationSchema.java:121)
at org.apache.flink.formats.json.JsonRowDataDeserializationSchema.deserialize(JsonRowDataDeserializationSchema.java:106)
... 9 more
Re: flink-1.13.1 sql error
Posted by Jingsong Li <ji...@gmail.com>.
你这个报错信息，意思是有个字段，你在DDL里声明是ROW，但是真实数据的这列却不是一个ROW。
你应该排查下是否声明对了，如果在期望中，可以通过ignoreParseErrors来跳过这行
Best,
Jingsong
On Mon, Jun 21, 2021 at 11:46 AM zhuxiaoshang <zh...@gmail.com>
wrote:
> Json反序列化失败了，像是数据问题
> {\n \"app_time\": \"2021-06-14 10:00:00\",\n
> \"category_id\": 1,\n \"item_id\": 1,\n
> \"user_id\": 1,\n \"behavior\": \"pv\"\n}
>
> > 2021年6月20日 下午9:08，kcz <57...@qq.com.INVALID> 写道：
> >
> > 大佬们 帮看下这个是为什么提示那个错误
> >
> >
> >
> > --
> > Sent from: http://apache-flink.147419.n8.nabble.com/
>
>
--
Best, Jingsong Lee
回复： flink-1.13.1 sql error
Posted by guoyb <86...@qq.com.INVALID>.
解析错误了吧，数据和定义的格式不对应。
------------------ 原始邮件 ------------------
发件人: "user-zh" <zhushang7841@gmail.com>;
发送时间: 2021年6月21日(星期一) 中午11:45
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: Re: flink-1.13.1 sql error
Json反序列化失败了，像是数据问题
{\n&nbsp; &nbsp; \"app_time\": \"2021-06-14 10:00:00\",\n&nbsp; &nbsp; \"category_id\": 1,\n&nbsp; &nbsp; \"item_id\": 1,\n&nbsp; &nbsp; \"user_id\": 1,\n&nbsp; &nbsp; \"behavior\": \"pv\"\n}
> 2021年6月20日 下午9:08，kcz <573693104@qq.com.INVALID> 写道：
>
> 大佬们 帮看下这个是为什么提示那个错误
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink-1.13.1 sql error
Posted by zhuxiaoshang <zh...@gmail.com>.
Json反序列化失败了，像是数据问题
{\n \"app_time\": \"2021-06-14 10:00:00\",\n \"category_id\": 1,\n \"item_id\": 1,\n \"user_id\": 1,\n \"behavior\": \"pv\"\n}
> 2021年6月20日 下午9:08，kcz <57...@qq.com.INVALID> 写道：
>
> 大佬们 帮看下这个是为什么提示那个错误
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink-1.13.1 sql error
Posted by kcz <57...@qq.com.INVALID>.
大佬们 帮看下这个是为什么提示那个错误
--
Sent from: http://apache-flink.147419.n8.nabble.com/