I Flink You Freaky And I like you a lot! 1.java

FlinkPipelineOptions pipelineOptions =PipelineOptionsFactory.create().as(FlinkPipelineOptions.class);
pipelineOptions.setJobName(jobName);
pipelineOptions.setRunner(FlinkRunner.class);
pipelineOptions.setParallelism(1);
pipelineOptions.setStreaming(false);
Pipeline p = Pipeline.create(pipelineOptions);

p.apply(Create.of("To be, or not to be: that is the question:",
        "Whether 'tis nobler in the mind to suffer", "The slings and arrows of fortune,",
        "Or to take arms against a sea of troubles,"))

        .apply(ParDo.of(new DoFn<String, String>() {
            private static final long serialVersionUID = 1;

            @ProcessElement
            public void processElement(ProcessContext c) {
                String line = c.element();
                for (int i = 0; i < 1000; i++) {
                    c.output(line);
                }
            }
        }))

        .apply(ParDo.of(new DoFn<String, String>() {
            private static final long serialVersionUID = 1;

            @ProcessElement
            public void processElement(ProcessContext c) {
                String[] words = c.element().split("\\s");
                for (String word : words) {
                    c.output(word);
                }
            }
        }))

        .apply(MapElements.into(TypeDescriptors.strings()).via(word -> {
            LOG.info("{}", word);
            return word;
        }));

p.run();