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

[Jingsong Li]

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 郑斌斌 <[hidden email]> 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

[郑斌斌]

谢谢了，查了下jira, No.1在1.11中才用修复, 另外，目前我用的版本就是1.10
https://issues.apache.org/jira/browse/FLINK-15396
------------------------------------------------------------------
发件人：Jingsong Li <[hidden email]>
发送时间：2020年7月3日(星期五) 14:29
收件人：user-zh <[hidden email]>; 郑斌斌 <[hidden email]>
主　题：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 郑斌斌 <[hidden email]> 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