flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "sean.miao (JIRA)" <j...@apache.org>
Subject [jira] [Created] (FLINK-10119) 存在数据非json格式,使用KafkaJsonTableSource的话,job无法拉起。
Date Fri, 10 Aug 2018 02:37:00 GMT
sean.miao created FLINK-10119:
---------------------------------

             Summary: 存在数据非json格式,使用KafkaJsonTableSource的话,job无法拉起。
                 Key: FLINK-10119
                 URL: https://issues.apache.org/jira/browse/FLINK-10119
             Project: Flink
          Issue Type: Bug
          Components: Kafka Connector
    Affects Versions: 1.5.1
         Environment: 无
            Reporter: sean.miao


开启checkpoint和savepoint,同时开启了job的自动拉起。

flink从kafka消费数据,使用的是Kafka010JsonTableSource。发现只要有一条数据非json格式,就会导致应用挂掉无法拉起。

当前,这仅是满足了处理语义,但是导致应用不可以用就不太好了吧。能不能改成像spark
sql一样,不满足格式的数据,增加到一个专门存储无法解析的数据的列里面。

 

我们目前的做法是

JsonRowDeserializationSchema

@Override
public Row deserialize(byte[] message) throws IOException {
 try {
 final JsonNode root = objectMapper.readTree(message);
 return convertRow(root, (RowTypeInfo) typeInfo);
 } catch (Throwable t) {
 throw new IOException("Failed to deserialize JSON object.", t);
 }
}

catch 里抛异常改成了传入一个 “{}”,会使得所有不能解析数据给所有列返回空值。



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message