Dataflow primitive operations

ExtractMessage

    // .apply("Extract Body", ParDo.of(new ExtractMessage()))
    class ExtractMessage extends DoFn<PubsubMessage, String> {
      @ProcessElement
      public void processElement(final ProcessContext c) {
        final PubsubMessage msg = c.element();
        final String s = new String(msg.getPayload());
        c.output(s);
      }
    }

Windowed GCS writes

        String targetPath "gs:bucket/subfolder/"
        p.apply(
            "Write Body",
            TextIO.write()
                .to(targetPath + "data-enriched-")
                .withWindowedWrites()
                .withNumShards(10)
                .withSuffix(".out"));