private final String PROJECT = "beer-coding";
private final String BUCKET = "gs://DataFlow-byjw";

DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
options.setRunner(DataflowPipelineRunner.class);
options.setProject(PROJECT);
options.setStagingLocation(BUCKET);
options.setStreaming(true);

Pipeline pipeline = Pipeline.create(options);

public static void main(String[] args) {
    Pipeline pipeline = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());
}

class MyFunction extends DoFn<{입력 데이타 타입},{출력 데이타 타입}> {
      @Override
      public void processElement(ProcessContext processContext) {
        String result = processContext.element() // 입력 데이터 타입으로 정의된 입력 데이터를 꺼냄
        processContext.output(result); // 파이프 라인상의 다음 컴포넌트로 데이터 전달
      }
}

Pipeline pipeline = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());

pipeline.apply(ParDo.named("toUpper").of(new DoFn<String,String>() {
    @Override
    public void processElement(ProcessContext processContext) {
        String result = processContext.element(); // 입력 데이터 타입으로 정의된 입력 데이터를 꺼냄
        processContext.output(result); // 파이프 라인상의 다음 컴포넌트로 데이터 전달
    }
}))

Pipeline pipeline = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());
pipeline.apply(ParDo.named("toUpper").of(new UpperCase());

class UpperCase extends DoFn<String, String> {
    @Override
    public void processElement(ProcessContext processContext) {
        processContext.output(processContext.element().toUpperCase());
    }
}

PCollection <KV<String, Iterable<Integer>>> groupedWords =
    wordAndLines.apply(GroupByKey.<String,Integer> create());

PCollection <KV<String, Iterable<Integer>>> groupedWords =
    wordAndLines.apply(GroupByKey.<String,Integer> create());
    
PCollection <KV<String, Integer>> sum = groupedWords.apply(Combine.globally(new Sum.SumIntegerFn)));
    

PCollection<String> items = ...;
PCollection<String> fixed_windowed_items = items.apply(
Window.<String>into(FixedWindows.of(1, TimeUnit.MINUTES)));

PCollection<String> items = ...;
  PCollection<String> sliding_windowed_items = items.apply(    Window.<String>into(SlidingWindows.of(Duration.standardMinutes(30)).every(Duration.standardSeconds(5))));

userEvents
  .apply(Window.named("WindowIntoSessions")

        .<KV<String, Integer>>into(

              Sessions.withGapDuration(Duration.standardMinutes(Duration.standardMinutes(10))))

    .withOutputTimeFn(OutputTimeFns.outputAtEndOfWindow()))

  // For this use, we care only about the existence of the session, not any particular

  // information aggregated over it, so the following is an efficient way to do that.

  .apply(Combine.perKey(x -> 0))

  // Get the duration per session.

  .apply("UserSessionActivity", ParDo.of(new UserSessionInfoFn()))