从PubSub读写Google Cloud Storage的数据流作业耗尽时的数据丢失

Question

当将固定数量的字符串（用于测试的800,000个1KB）放入PubSub主题并在Dataflow中运行以下Apache Beam(2.1.0)作业时，只要按照预期保留了语义。

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

public class PubSubToGsSimpleJob {

    public static void main(String[] args) {
        PubSubToGsPipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                .as(PubSubToGsPipelineOptions.class);
        Pipeline p = Pipeline.create(options);

        p.apply(PubsubIO.readStrings().fromSubscription(options.getInput()))
                .apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))))
                .apply(TextIO.write().withWindowedWrites().withNumShards(1).to(options.getOutput()));
        p.run();
    }

}

import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;

public interface PubSubToGsPipelineOptions extends PipelineOptions {
    @Description("PubSub subscription")
    String getInput();
    void setInput(String input);

    @Description("Google Cloud Storage output path")
    String getOutput();
    void setOutput(String output);
}

Answer 1

我的猜测是，在drain and replacement job用剩余窗口覆盖某个窗口之前，该窗口可能已经被部分写入。您可以在WriteFiles中检查已耗尽作业和替换作业中的工作日志。如果使用波束头，当最终目标被覆盖时，它也会记录下来。

从概念上讲，耗尽作业和替换作业是完全不同的管道。使用相同的输出位置与对其他两个不相关的作业使用相同的输出位置没有什么不同。您可以尝试的另一件事是为第二个作业使用不同的输出路径，并验证所有记录都存在于两个目录中。