Simple side output

gistfile1.java

Pipeline p = Pipeline.create(options);

        final TupleTag<String> output1 = new TupleTag<String>(){};
        final TupleTag<String> output2 = new TupleTag<String>(){};

        PCollectionTuple tuples = p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
         .apply(ParDo
        		 .withOutputTags(output1, TupleTagList.of(output2))
        		 .named("ExtractWords").of(new DoFn<String, String>() {
                           private static final long serialVersionUID = 0;
                           @Override
                           public void processElement(ProcessContext c) {
                             for (String word : c.element().split("[^a-zA-Z']+")) {
                               if (!word.isEmpty()) {
                                 c.output(word);
                               } else {
                            	   c.sideOutput(output2, word);
                               }
                             }
                           }
                         }));
        tuples.get(output1).apply(Count.<String>perElement())
         .apply(ParDo.named("FormatResults").of(new DoFn<KV<String, Long>, String>() {
                           private static final long serialVersionUID = 0;
                           @Override
                           public void processElement(ProcessContext c) {
                             c.output(c.element().getKey() + ": " + c.element().getValue());
                           }
                         }))
         .apply(TextIO.Write.to("gs://sj_test/df-output/"));

        tuples.get(output2).apply(Count.<String>perElement())
        .apply(ParDo.named("FormatResults2").of(new DoFn<KV<String, Long>, String>() {
                          private static final long serialVersionUID = 0;
                          @Override
                          public void processElement(ProcessContext c) {
                            c.output(c.element().getKey() + ": " + c.element().getValue());
                          }
                        }))
        .apply(TextIO.Write.to("gs://sj_test/df-output/"));


        p.run();