flink sql如何处理脏数据问题？

[casel.chen]

场景：使用flink 
sql将数据写入下游OLAP系统，如doris，遇到一些异常情况，比如字段值超长或者分区字段值为当前doris表不存在的分区（需要先人为创建）等等，当前写入这些脏数据会使得作业写入报错，进而导致作业失败。我们是希望能够将这些“脏”数据单独发到一个kafka
 topic或者写入一个文件便于事后审查。这个目前有办法做到吗？