如何解决S3的Apache Flink状态读取文件
我有一个flink批处理作业，该作业从S3中读取一个非常大的镶木文件，然后将json存储到Kafka主题中。
问题是如何使文件读取过程有状态？我的意思是，每当工作中断或被破坏时，工作应从先前的阅读状态开始吗？作业重新启动时，我不想将重复的项目发送给Kafka。
这是我的示例代码
val env = ExecutionEnvironment.getExecutionEnvironment
val input = Parquet.input[User](new Path(s"s3a://path"))
env.createInput(input)
.filter(r => Option(r.token).getorElse("").nonEmpty)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
