在Flink中解析JSON时如何处理异常

Question

我使用flink 1.4.2从Kafka读取数据，并使用JSONDeserializationSchema将它们解析为ObjectNode。如果传入的记录不是有效的JSON，那么我的Flink作业就会失败。我不想让这项工作失败，而想跳过这个破纪录。

FlinkKafkaConsumer010<ObjectNode> kafkaConsumer =
new FlinkKafkaConsumer010<>(TOPIC, new JSONDeserializationSchema(), consumerProperties);
DataStream<ObjectNode> messageStream = env.addSource(kafkaConsumer);
messageStream.print();

如果Kafka中的数据不是有效的JSON，我会得到以下异常。

Job execution switched to status FAILING.
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'This': was expecting ('true', 'false' or 'null')
at [Source: [B@4f522623; line: 1, column: 6]
Job execution switched to status FAILED.
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.

Answer 1

最简单的解决方案是实现自己的DeserializationSchema并包装JSONDeserializationSchema。然后，您可以捕获异常并忽略它或执行自定义操作。

Answer 2

根据@twathr的建议，我通过复制JSONDeserializationSchema实现了自己的DeserializationSchema，并添加了异常处理。

import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;
public class CustomJSONDeserializationSchema extends AbstractDeserializationSchema<ObjectNode> {
private ObjectMapper mapper;
@Override
public ObjectNode deserialize(byte[] message) throws IOException {
if (mapper == null) {
mapper = new ObjectMapper();
}
ObjectNode objectNode;
try {
objectNode = mapper.readValue(message, ObjectNode.class);
} catch (Exception e) {
ObjectMapper errorMapper = new ObjectMapper();
ObjectNode errorObjectNode = errorMapper.createObjectNode();
errorObjectNode.put("jsonParseError", new String(message));
objectNode = errorObjectNode;
}
return objectNode;
}
@Override
public boolean isEndOfStream(ObjectNode nextElement) {
return false;
}
}

在我的流媒体工作中

messageStream
.filter((event) -> {
if(event.has("jsonParseError")) {
LOG.warn("JsonParseException was handled: " + event.get("jsonParseError").asText());
return false;
}
return true;
}).print();

Answer 3

Flink改进了FlinkKafkaConsumer 的空记录处理

当DeserializationSchema遇到损坏的消息时，有两种可能的设计选择。它可以抛出一个导致管道重新启动的IOException，也可以返回null，Flink Kafka使用者将静默地跳过损坏的消息。

有关更多详细信息，您可以查看此链接。