如何正确处理自定义MapFunction中的错误？

Question

我已经实现MapFunction了我的Apache Flink流程.它正在解析传入的元素并将它们转换为其他格式,但有时会出现错误(即传入的数据无效).

我看到了两种可能的处理方式:

• 忽略无效元素,但似乎我不能忽略错误,因为对于任何传入元素我必须提供传出元素.
• 将传入的元素拆分为有效和无效但似乎我应该使用其他功能.

所以,我有两个问题:

1. 如何正确处理我的错误MapFunction？
2. 如何正确实现这样的转换功能？

Answer 1

你可以使用a FlatMapFunction而不是a MapFunction.这将允许您仅在元素有效时才发出元素.以下显示了一个示例实现:

input.flatMap(new FlatMapFunction<String, Long>() {
    @Override
    public void flatMap(String input, Collector<Long> collector) throws Exception {
        try {
            Long value = Long.parseLong(input);
            collector.collect(value);
        } catch (NumberFormatException e) {
            // ignore invalid data
        }
    }
});