Apache Beam Kinesio Java-使用kinesis流中的数据

Question

首先，我想说这对Beam世界来说是全新的。我正在处理一个以Apache Beam为中心的任务，我的主要数据源是Kinesis流。在那里，当我使用流数据时，我注意到当我重新启动程序（我的消费者应用程序）时，相同的数据集也会出现。这是我的代码，

    String awsStreamName = KinesisStream.getProperty("stream.name");
    String awsAccessKey = KinesisStream.getProperty("access.key");
    String awsSecretKey = KinesisStream.getProperty("secret.key");
    String awsRegion = KinesisStream.getProperty("aws.region");
    Regions region = Regions.fromName(awsRegion);

    return KinesisIO.read()
            .withStreamName(awsStreamName)
            .withInitialPositionInStream(InitialPositionInStream.LATEST)
            .withAWSClientsProvider(awsAccessKey, awsSecretKey, region);

简单地说，我想要的是，我需要从我之前阅读的地方开始阅读数据。如果有人能提供一些资源，我真的很感激。

我还发现了一个类似的问题，但它对我没有帮助-Apache Beam Kinesis IOJava处理管道-应用程序状态，错误处理

Answer 1

波束中的无界震源，如Kinesio。read（）支持在重新启动应用程序后使用检查点标记从最新的检查点恢复检查点。

这些检查点必须持久化到持久存储。但是，具体如何完成取决于您正在使用的Beam运行程序，例如Dataflow、Apache Flink或Apache Spark。

我建议在检查点上阅读相应运行时的留档，并检查相应Beam运行程序的管道选项。

例如，在Apache Flink的情况下，您必须通过check point ingInterval（Flink PipelineOptions）启用检查点，并在Flink中配置检查点。