title: Apache Beam Streaming pipeline with sequential batches
body: <p><strong>What I am trying to do:</strong></p>
  <ol>
  <li><p>Consume json messages from PubSub subscription using Apache Beam Streaming pipeline &amp; Dataflow Runner</p>
  </li>
  <li><p>Unmarshal payload strings into objects.</p>
  <ul>
  <li>Assume 'messageId' is the unique Id of incoming message. Ex: msgid1, msgid2, etc</li>
  </ul>
  </li>
  <li><p>Retrieve child records from a database for each object resulted from #2. Same child can be applicable for multiple messages.</p>
  <ul>
  <li>Assume 'childId' as the unique Id of child record. Ex: cid1234, cid1235 etc</li>
  </ul>
  </li>
  <li><p>Group child records by their unique id as shown in example below</p>
  <ul>
  <li>KV.of(cid1234,Map.of(msgid1, msgid2)) and KV.of(cid1235,Map.of(msgid1, msgid2))</li>
  </ul>
  </li>
  <li><p>Write grouped result at childId level to the database</p>
  </li>
  </ol>
  <p><strong>Questions:</strong></p>
  <ol>
  <li><p>Where should the windowing be introduced? we currently have 30minutes fixed windowing after step#1</p>
  </li>
  <li><p>How does Beam define start and end time of 30mins window? is it right after we start pipeline or after first message of batch?</p>
  </li>
  <li><p>What if the steps 2 to 5 take more than 1hour for a window and next window batch is ready. Would both windows batches gets processed in parallel?</p>
  </li>
  <li><p>How can make the next window messages wait until previous window batch is completed?</p>
  <ul>
  <li>If we dont do this, the result at childId level will be overwritten by next batches</li>
  </ul>
  </li>
  </ol>
  <p><strong>Code snippet:</strong></p>
  <pre><code>         PCollection&lt;PubsubMessage&gt; messages = pipeline.apply(&quot;ReadPubSubSubscription&quot;,
               PubsubIO.readMessagesWithAttributes()
                   .fromSubscription(&quot;projects/project1/subscriptions/subscription1&quot;));
  
           PCollection&lt;PubsubMessage&gt; windowedMessages = messages.apply(Window.into(FixedWindows
               .of(Duration.standardMinutes(30))));
               
           PCollectionTuple unmarshalResultTuple = windowedMessages.apply(&quot;UnmarshalJsonStrings&quot;,
               ParDo.of(new JsonUnmarshallFn())
                   .withOutputTags(JsonUnmarshallFn.mainOutputTag,
                       TupleTagList.of(JsonUnmarshallFn.deadLetterTag)));
  
           PCollectionTuple childRecordsTuple = unmarshalResultTuple
               .get(JsonUnmarshallFn.mainOutputTag)
               .apply(&quot;FetchChildsFromDBAndProcess&quot;,
                   ParDo.of(new ChildsReadFn() )
                       .withOutputTags(ChildsReadFn.mainOutputTag,
                           TupleTagList.of(ChildsReadFn.deadLetterTag)));
  
           // input is KV of (childId, msgids), output is mutations to write to BT
           PCollectionTuple postProcessTuple = childRecordsTuple
               .get(ChildsReadFn.mainOutputTag)
               .apply(GroupByKey.create())
               .apply(&quot;UpdateChildAssociations&quot;,
                   ParDo.of(new ChildsProcessorFn())
                       .withOutputTags(ChildsProcessorFn.mutations,
                           TupleTagList.of(ChildsProcessorFn.deadLetterTag)));
  
           postProcessTuple.get(ChildsProcessorFn.mutations).CloudBigtableIO.write(...);
  </code></pre>