very-minimal-beam-pipeline.java

package sungjunyoung.github.io;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.TextualIntegerCoder;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;

import java.util.ArrayList;


public class TestPipeline {

	static class ComputeWordLengthFn extends DoFn<String, Integer> {
		@ProcessElement
		public void processElement(ProcessContext c){
			String word = c.element();
			c.output(word.length());
		}
	}

	public static void main(String[] args) {

		PipelineOptions options = PipelineOptionsFactory.create();
		Pipeline p = Pipeline.create(options);

		ArrayList<String> temp = new ArrayList<String>();
		temp.add("안녕하세요");
		temp.add("저는");
		temp.add("성준영입니다.");
		temp.add("경희대학교 컴퓨터공학과에");
		temp.add("3학년으로 재학중입니다.");

		PCollection<String> test = p.apply(Create.of(temp));
		PCollection<Integer> testLength = test.apply(ParDo.of(new ComputeWordLengthFn()));

		testLength.apply("WriteNumbers",TextIO.Write.to("test-output").withSuffix(".txt").withCoder(TextualIntegerCoder.of()));
		p.run();
	}
}