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 郑斌斌 <zh...@heint.cn> on 2020/07/03 05:38:10 UTC
kafkaf To mysql 写入问题
dear:
请教两个问题
1) 用下面的代码消费kafka 发生序列化异常时,会发生JOB反复重试,重启后也是这样,
改用FlinkKafkaConsumer010类的话,有相关的解决方法,参照https://stackoverflow.com/questions/51301549/how-to-handle-exception-while-parsing-json-in-flink/51302225
不知道,用Kafka类的话,如何解决
.connect(
new Kafka()
.version("0.10")
.topic("test-input")
2) 对于timestamp类型字段,用JDBCAppendTableSink 把DataStream<Row>写入到mysql时,会发下面的错误LocalTimeStamp到Timestamp的转型错误
kafka消息是avro格式,字段类型设置为timestamp(3),我是把System.currentTimeMillis()写入到kafka中的
jdbc参数类型设置为Types.SQL_TIMESTAMP
thanks
回复:kafkaf To mysql 写入问题
Posted by 郑斌斌 <zh...@heint.cn>.
谢谢了,查了下jira, No.1在1.11中才用修复, 另外,目前我用的版本就是1.10
https://issues.apache.org/jira/browse/FLINK-15396
------------------------------------------------------------------
发件人:Jingsong Li <ji...@gmail.com>
发送时间:2020年7月3日(星期五) 14:29
收件人:user-zh <us...@flink.apache.org>; 郑斌斌 <zh...@heint.cn>
主 题:Re: kafkaf To mysql 写入问题
Hi,
估计需要使用Flink 1.11。
1.JSON Format有参数控制 [1]
2.是之前的bug,Flink 1.11应该是不会存在了,不确定1.10.1有没有修。
[1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/formats/json.html#json-ignore-parse-errors
Best,
Jingsong
On Fri, Jul 3, 2020 at 1:38 PM 郑斌斌 <zh...@heint.cn> wrote:
dear:
请教两个问题
1) 用下面的代码消费kafka 发生序列化异常时,会发生JOB反复重试,重启后也是这样,
改用FlinkKafkaConsumer010类的话,有相关的解决方法,参照https://stackoverflow.com/questions/51301549/how-to-handle-exception-while-parsing-json-in-flink/51302225
不知道,用Kafka类的话,如何解决
.connect(
new Kafka()
.version("0.10")
.topic("test-input")
2) 对于timestamp类型字段,用JDBCAppendTableSink 把DataStream<Row>写入到mysql时,会发下面的错误LocalTimeStamp到Timestamp的转型错误
kafka消息是avro格式,字段类型设置为timestamp(3),我是把System.currentTimeMillis()写入到kafka中的
jdbc参数类型设置为Types.SQL_TIMESTAMP
thanks
--
Best, Jingsong Lee
Re: kafkaf To mysql 写入问题
Posted by Jingsong Li <ji...@gmail.com>.
Hi,
估计需要使用Flink 1.11。
1.JSON Format有参数控制 [1]
2.是之前的bug,Flink 1.11应该是不会存在了,不确定1.10.1有没有修。
[1]
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/formats/json.html#json-ignore-parse-errors
Best,
Jingsong
On Fri, Jul 3, 2020 at 1:38 PM 郑斌斌 <zh...@heint.cn> wrote:
> dear:
> 请教两个问题
> 1) 用下面的代码消费kafka 发生序列化异常时,会发生JOB反复重试,重启后也是这样,
> 改用FlinkKafkaConsumer010类的话,有相关的解决方法,参照
> https://stackoverflow.com/questions/51301549/how-to-handle-exception-while-parsing-json-in-flink/51302225
> 不知道,用Kafka类的话,如何解决
> .connect(
> new Kafka()
> .version("0.10")
> .topic("test-input")
> 2) 对于timestamp类型字段,用JDBCAppendTableSink
> 把DataStream<Row>写入到mysql时,会发下面的错误LocalTimeStamp到Timestamp的转型错误
>
> kafka消息是avro格式,字段类型设置为timestamp(3),我是把System.currentTimeMillis()写入到kafka中的
> jdbc参数类型设置为Types.SQL_TIMESTAMP
> thanks
>
--
Best, Jingsong Lee