Hi Chao I think side output [1] might meet your requirements.
[1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html Best Yun Tang ________________________________ From: 范超 <fanc...@mgtv.com> Sent: Tuesday, August 25, 2020 10:54 To: user <user@flink.apache.org> Subject: How to sink invalid data from flatmap Hi, I’m using the custom flatmap function to validate the kafka json string message, if the kafka message is valid to transform to a pojo (using GSON), then go on with the next sink step. If it can not be parsed as a POJO, the GSON will throw the “com.google.gson.JsonSyntaxException”, and in my custom flatmap function , I just catch this exception, and then go on, but this invalidated json message is just omitted. But now, I want to save the invalidated json message to sink the original kafka json string to another table, but don’t know how to implement in my custom flatmap function, because the richmapfucntion has limited the collect type. Could someone give me some advice please? Thanks in advance! Chao Fan
