PubSub에서 읽고 데이터를 Google Cloud Storage에 쓰는 데이터 흐름 작업

Question

고정 된 수의 문자열 (테스트에 사용 된 800,000 개의 1KB)을 PubSub 항목에 넣고 다음 Apache Beam (2.1.0) 작업을에서 실행하는 경우 정확히 한 번 데이터 흐름은 예상대로 보존됩니다. (데이터 흐름 콘솔에서와 같이), 그리고 다시 출력 파일을 쫓겨 같은 작업이 실행되는 경우, 모든 요소 전에 배수

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); 
} 

은 읽기

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(); 
    } 

} 

PipelineOptions 구현 이하 PubSub 주제에 공개 된 원래 데이터 세트보다 적은 수의 레코드가 있습니다. 작업을 배수하고 교체하면 데이터 손실이 발생할 수 있음을 알 수 있습니다. 이는 this google cloud blog post에 적어도 에 의미가 있어야한다고 언급하기 때문에 이상합니다. 이 파이프 라인은 작업을 배수하고 교체 할 때 적어도 한 번 이상의 의미 (또는 한 번 더 정확하게 한 번 의미론)를 달성하도록 설계되어야합니까?

Answer 1

제 생각 엔 드레인 및 교체 작업이 나머지 창으로 덮어 쓰기 전에 창이 부분적으로 작성 될 수 있습니다. 이 로그 라인의 배수 작업 및 교체 작업에서 작업자 로그를 확인할 수 있습니다 in WriteFiles. Beam HEAD를 사용하면 최종 목적지를 덮어 쓸 때 기록됩니다.

개념적으로 배수 된 작업 및 교체 작업은 완전히 다른 파이프 라인입니다. 동일한 출력 위치를 사용하는 것이 다른 관련이없는 두 개의 작업에 동일한 출력 위치를 사용하는 것과 다르지 않습니다. 시도 할 수있는 또 다른 일은 두 번째 작업에 대해 다른 출력 경로를 사용하고 두 레코드 모두에 모든 레코드가 존재하는지 확인하는 것입니다.