Apache Flink错误处理和条件处理

Question

我是Flink的新手，已经通过网站/示例/博客开始学习。我正在努力正确使用操作符。基本上我有两个问题

问题1：Flink是否支持声明性异常处理，我需要处理解析/验证/。。。错误？

• 我可以使用组织吗。阿帕奇。Flink。运行时。操作员。分类ExceptionHandler或类似的程序来处理错误
• 还是Rich/FlatMap功能是我的最佳选择？如果Rich/FlatMap是唯一的选项，那么是否有办法在Rich/FlatMap函数中获取流的句柄，以便连接接收器进行错误处理

问题2：我可以有条件地附加不同的水槽吗？

• 根据键控分割流中的特定字段，我需要选择不同的接收器，我是再次分割流还是使用富/平面图来处理

我正在使用Flink 1.3.2。这是我工作的相关部分

    .....
    .....
    DataStream<String> eventTextStream = env.addSource(messageSource)

    KeyedStream<EventPojo, Tuple> eventPojoStream = eventTextStream
            // parse, transform or enrich
            .flatMap(new MyParseTransformEnrichFunction())
            .assignTimestampsAndWatermarks(new EventAscendingTimestampExtractor())
            .keyBy("eventId");

    // split stream based on eventType as different reduce and windowing functions need to be applied
    SplitStream<EventPojo> splitStream = eventPojoStream
            .split(new EventStreamSplitFunction());

    // need to apply reduce function
    DataStream<EventPojo> event1TypeStream = splitStream.select("event1Type");

    // need to apply reduce function
    DataStream<EventPojo> event2TypeStream = splitStream.select("event2Type");

    // need to apply time based windowing function
    DataStream<EventPojo> event3TypeStream = splitStream.select("event3Type");

    ....
    ....

    env.execute("Event Processing");      

我在这里使用的运算符是否正确？

更新1：

尝试使用@alpinegizmo建议的ProcessFunction，但这不起作用，因为它依赖于一个键控流，而我在解析/验证输入之前没有键控流。我得到“InvalidProgrameException：对于非复合类型，字段表达式必须等于'*'或'\u''”。

这是一个常见的用例，您的第一次解析/验证输入，但还没有键控流，所以如何解决它？

谢谢你的耐心和帮助。

Answer 1

您忽略了一个关键的构建块。查看侧面输出。

此机制提供了一种类型安全的方式来生成任意数量的附加输出流。这可能是报告错误的干净方法，以及其他用途。在Flink中，1.3侧输出只能与ProcessFunction一起使用，但1.4会将侧输出添加到ProcessWindowFunction。