flink sql如何处理脏数据？

业务上游数据源发出来的数据有可能会有脏数据导致数据无法解析成源表的结构，如kafka json topic映射成源表。
请问这种情况下flink sql要如何处理? 期望的是将脏数据发到一个专门的topic，是不是要自己写个connector? 标准kafka 
connector支持这种需求么？

回复