flink中的排重逻辑如何处理比较好

flink中如何对于所有的数据进行基于主键的去重。目前采用的keyBy(pk)后通过状态过滤，实际运行的时候checkpoints时间太长，非常阻塞性能。有没有什么更好的处理方式。另外flink-sql中的distinct具体实现逻辑是什么样子，必须指定窗口么。