title: Write to Cloud Storage every X messages from Pubsub
body: <p>I am new to Cloud Dataflow / Apache Beam, so the concept/programming is still hazy to me.</p>
  
  <p>What I want to do is that Dataflow listens to Pubsub and gets messages of this format in JSON:</p>
  
  <pre><code>{
    "productId": "...",
    "productName": "..."
  }
  </code></pre>
  
  <p>And transform that to:</p>
  
  <pre><code>{
    "productId": "...",
    "productName": "...",
    "sku": "...",
    "inventory": {
      "revenue": &lt;some Double&gt;,
      "stocks":  &lt;some Integer&gt;
    }
  }
  </code></pre>
  
  <p>So the steps needed are:</p>
  
  <ol>
  <li><p>(<em>IngestFromPubsub</em>) Get records from Pubsub by listening to a topic (1 Pubsub message = 1 record)</p></li>
  <li><p>(<em>EnrichDataFromAPI</em>)</p>
  
  <p>a. Deserialize the payload's JSON string into Java object</p>
  
  <p>b. By calling an external API, using the <code>sku</code>, I can enrich the data of each record by adding the <code>inventory</code> attribute.</p>
  
  <p>c. Serialize the records again.</p></li>
  <li><p>(<em>WriteToGCS</em>) Then every <code>x</code> number (can be parameterized) records, I need to write these in Cloud Storage. 
  Please consider also the trivial case that <code>x=1</code>. 
  (Does <code>x=1</code>, a good idea? I am afraid there will be too many Cloud Storage writes)</p></li>
  </ol>
  
  <p>Even though I am a Python guy, I am already having difficulty doing this in Python, more so that I need to do write in Java. I am getting headache reading Beam's example in Java, it's too verbose and difficult to follow. All I understand is that each step is an <code>.apply</code> to the PCollection.</p>
  
  <p>So far, here is the result of my puny effort:</p>
  
  <pre><code>public static void main(String[] args) {
  
      Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
      options.setStreaming(true);
  
      Pipeline pipeline = Pipeline.create(options);
      pipeline
          .apply("IngestFromPubsub", PubsubIO.readStrings().fromTopic(options.getTopic()))
          // I don't really understand the next part, I just copied from official documentation and filled in some values
          .apply(Window.&lt;String&gt;into(FixedWindows.of(Duration.millis(5000)))
              .withAllowedLateness(Duration.millis(5000))
              .triggering(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(1000)))
              .discardingFiredPanes()
          )
          .apply("EnrichDataFromAPI", ParDo.of(
              new DoFn&lt;String, String&gt;() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                      c.element();
                      // help on this part, I heard I need to use Jackson but I don't know, for API HttpClient is sufficient
                      // ... deserialize, call API, serialize again ...
                      c.output(enrichedJSONString);
                  }
              }
          ))
          .apply("WriteToGCS", 
              TextIO.write().withWindowedWrites().withNumShards(1).to(options.getOutput()))
      ;
  
  
      PipelineResult result = pipeline.run();
  }
  </code></pre>
  
  <p>Please fill in the missing parts, and also give me a tip on Windowing (e.g. what's the appropriate configuration etc.) and in which steps should I insert/apply it.</p>