final String MY_PROJECT = "by-jungwoon"; // 본인의 GCP 프로젝트 ID를 입력
final String MY_GCS_FOR_STAGING = "gs://by-jungwoon"; // 본인의 GCS 버킷 주소를 입력

// Google Dataflow 관련 내 정보를 가진 옵션을 생성
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setProject(MY_PROJECT);
options.setStagingLocation(MY_GCS_FOR_STAGING);
options.setRunner(DataflowRunner.class);

// 위의 옵션으로 파이프라인 생성
Pipeline pipeline = Pipeline.create(options);

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

PCollection<String> lines = pipeline.apply(
      "ReadMyFile", TextIO.read().from("protocol://path/to/some/inputData.txt"));

PCollection<데이터 타입> collection = [PipelineI/O or Transforms로 생성]

[Output PCollection] = [Input PCollection].apply([Transform])

[Output PCollection 1] = [Input PCollection].apply([Transform 1])
[Output PCollection 2] = [Input PCollection].apply([Transform 2])

Pipeline pipeline = Pipeline.create(options);

pipeline.apply("Transform 이름", ParDo.of(
    new DoFn<입력 데이타 타입 , 출력 데이타 타입>() {
        @ProcessElement
        public void processElement(ProcessContext processContext) {
            String result = processContext.element(); // 입력 데이터 타입으로 정의된 입력 데이터를 꺼냄
            ... // 여기서 데이터들의 처리를 합니다.
            processContext.output(result); // 파이프 라인상의 다음 컴포넌트로 데이터 전달
        }
    }
))

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

// 여기서 MyFunction은 위에서 만들어놓은 DoFn(=MyFunction()) 입니다.
pipeline.apply("toUpper", ParDo.of(new MyFunction());

public static void main(String[] args) {

        // Google Dataflow 관련 내 정보를 가진 옵션을 생성
        DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
        options.setProject("by-jungwoon");
        options.getGcpTempLocation();
        options.setRunner(DataflowRunner.class);

        // 위의 옵션으로 파이프라인 생성
        Pipeline pipeline = Pipeline.create(options);

        pipeline.apply(TextIO.read().from("gs://apache-beam-samples/shakespeare/kinglear.txt")) // 데이터 읽어옴
                .apply("Test", MapElements // 새로운 PTransform
                        .into(TypeDescriptors.strings()) // Output Type을 지정
                        .via((String word) -> word.toUpperCase())) // Input / Output Type을 지정, 대문자로 바꾼 결과
                .apply(TextIO.write().to("gs://by-jungwoon/hello.txt")); // 변경된 결과를 GCS에 저장

        pipeline.run().waitUntilFinish();
    }

apply("ExtractWords", FlatMapElements
        .into(TypeDescriptors.strings())
        .via((String word) -> Arrays.asList(word.split("[^\\p{L}]+"))))

List<KV<String, Integer>> wordAndNumber = Arrays.asList(
        KV.of("cat", 1),
        KV.of("dog", 5),
        KV.of("and", 1),
        KV.of("jump", 3),
        KV.of("tree", 2),
        KV.of("cat", 5),
        KV.of("dog", 2),
        KV.of("and", 2),
        KV.of("cat", 9),
        KV.of("and", 6)
);

PCollection<KV<String, Integer>> objects = pipeline.apply("CreateObjects", Create.of(wordAndNumber));
PCollection<KV<String, Iterable<Integer>>> groupedWords = objects.apply(GroupByKey.create());

// Key-Value 리스트1
final List<KV<String, String>> emailsList = Arrays.asList(
    KV.of("amy", "amy@example.com"),
    KV.of("carl", "carl@example.com"),
    KV.of("julia", "julia@example.com"),
    KV.of("carl", "carl@email.com"));

// Key-Value 리스트2
final List<KV<String, String>> phonesList = Arrays.asList(
    KV.of("amy", "111-222-3333"),
    KV.of("james", "222-333-4444"),
    KV.of("amy", "333-444-5555"),
    KV.of("carl", "444-555-6666"));

// 각각 PColection으로 형변환
PCollection<KV<String, String>> emails = p.apply("CreateEmails", Create.of(emailsList));
PCollection<KV<String, String>> phones = p.apply("CreatePhones", Create.of(phonesList));

// 각 리스트별로 Tag를 달아줌
final TupleTag<String> emailsTag = new TupleTag<>();
final TupleTag<String> phonesTag = new TupleTag<>();

// 실제 연산
PCollection<KV<String, CoGbkResult>> results =
    KeyedPCollectionTuple
    .of(emailsTag, emails)
    .and(phonesTag, phones)
    .apply(CoGroupByKey.create());

PCollection<String> contactLines = results.apply(ParDo.of(
  new DoFn<KV<String, CoGbkResult>, String>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
      KV<String, CoGbkResult> e = c.element();
      String name = e.getKey();
      Iterable<String> emailsIter = e.getValue().getAll(emailsTag);
      Iterable<String> phonesIter = e.getValue().getAll(phonesTag);
      String formattedResult = Snippets.formatCoGbkResults(name, emailsIter, phonesIter);
      c.output(formattedResult);
    }
  }
));

// Sum.SumIntegerFn() combines the elements in the input PCollection. The resulting PCollection, called sum,
// contains one value: the sum of all the elements in the input PCollection.
PCollection<Integer> pc = ...;
PCollection<Integer> sum = pc.apply(
   Combine.globally(new Sum.SumIntegerFn()));

public class AverageFn extends CombineFn<Integer, AverageFn.Accum, Double> {

  public static class Accum {
    int sum = 0;
    int count = 0;
  }

  @Override
  public Accum createAccumulator() { return new Accum(); }

  @Override
  public Accum addInput(Accum accum, Integer input) {
      accum.sum += input;
      accum.count++;
      return accum;
  }

  @Override
  public Accum mergeAccumulators(Iterable<Accum> accums) {
    Accum merged = createAccumulator();
    for (Accum accum : accums) {
      merged.sum += accum.sum;
      merged.count += accum.count;
    }
    return merged;
  }

  @Override
  public Double extractOutput(Accum accum) {
    return ((double) accum.sum) / accum.count;
  }
}

// Flatten takes a PCollectionList of PCollection objects of a given type.// Fla
// Returns a single PCollection that contains all of the elements in the PCollection objects in that list.
PCollection<String> pc1 = ...;
PCollection<String> pc2 = ...;
PCollection<String> pc3 = ...;
PCollectionList<String> collections = PCollectionList.of(pc1).and(pc2).and(pc3);

PCollection<String> merged = collections.apply(Flatten.<String>pCollections());

PCollection<Student> students = ...;
PCollectionList<Student> studentsByPercentile =
    students.apply(Partition.of(10, new PartitionFn<Student>() {
        public int partitionFor(Student student, int numPartitions) {
            return student.getPercentile()  // 0..99
                 * numPartitions / 100;
        }}));

// You can extract each partition from the PCollectionList using the get method, as follows:
PCollection<Student> fortiethPercentile = studentsByPercentile.get(4);

PCollection<String> items = ...;
PCollection<String> fixedWindowedItems = items.apply(
    Window.<String>into(FixedWindows
    .of(Duration.standardSeconds(30))));

PCollection<String> items = ...;
PCollection<String> slidingWindowedItems = items.apply(
    Window.<String>into(SlidingWindows
    .of(Duration.standardSeconds(60)) // 윈도우 길이 60초
    .every(Duration.standardSeconds(30)))); // 윈도우 간격 30초

PCollection<String> items = ...;
PCollection<String> sessionWindowedItems = items.apply(
    Window.<String>into(Sessions.withGapDuration(Duration.standardMinutes(10))));

PCollection<String> items = ...;
    PCollection<String> fixedWindowedItems = items.apply(
        Window.<String>into(FixedWindows.of(Duration.standardMinutes(1)))
              .withAllowedLateness(Duration.standardDays(2)));