title: Apache Beam : Transform an objects having a list of objects to multiple TableRows to write to BigQuery
body: <p>I am working on a beam pipeline to process a json and write it to bigquery. The JSON is like this. </p>
  
  <pre><code>{
  "message": [{
      "name": "abc",
      "itemId": "2123",
      "itemName": "test"
  
  }, {
      "name": "vfg",
      "itemId": "56457",
      "itemName": "Chicken"
  }],
  "publishDate": "2017-10-26T04:54:16.207Z"
  </code></pre>
  
  <p>}</p>
  
  <p>I parse this using Jackson to the below structure. </p>
  
  <pre><code>class Feed{
  List&lt;Message&gt; messages; 
  TimeStamp  publishDate;
  
  }
  </code></pre>
  
  <hr>
  
  <pre><code>public class Message implements Serializable{
  
  /**
   * 
   */
  private static final long serialVersionUID = 1L;
  private String key;
  private String value;
  
  private Map&lt;String, String&gt; eventItemMap = new HashMap&lt;&gt;();
  this property translate the list of map as a single map with all the key-value pair together. because,  the messages property will be parsed as list of HashMap objets for each key/value. This will be translated to a single map. 
  </code></pre>
  
  <p>Now in my pipeline, I will convert the collection as </p>
  
  <pre><code>PCollection&lt;KV&lt;String, Feed&gt;&gt;
  </code></pre>
  
  <p>to write it to different tables based on a property in the class. I have written a transform to do this. 
  The requirement is to create multiple TableRows based on the number of message objects. I have a few more properties in the JSON to along with publishDate which would be added to the tableRow and each message properties. 
  So the table would be as follows. </p>
  
  <pre><code>id, name, field1, field2, message1.property1, message1.property2...
  
  id, name, field1, field2, message2.property1, message2.property2...
  </code></pre>
  
  <p>I tried to create the below transformation. But, not sure how it will output multiple rows based on the message list. </p>
  
  <pre><code>private class BuildRowListFn extends DoFn&lt;KV&lt;String, Feed&gt;, List&lt;TableRow&gt;&gt; {
  
      @ProcessElement
      public void processElement(ProcessContext context) {
          Feed feed = context.element().getValue();
  
          List&lt;Message&gt; messages = feed.getMessage();
          List&lt;TableRow&gt; rows = new ArrayList&lt;&gt;();
          messages.forEach((message) -&gt; {
              TableRow row = new TableRow();
              row.set("column1", feed.getPublishDate());
              row.set("column2", message.getEventItemMap().get("key1"));
              row.set("column3", message.getEventItemMap().get("key2"));
              rows.add(row);
          }
  
          );
  
      }
  </code></pre>
  
  <p>But, this also will be a List which I won't be able to apply the BigQueryIO.write transformation. </p>
  
  <hr>
  
  <h2>Updated as per the comment from "Eugene" aka @jkff</h2>
  
  <p>Thanks @jkff. Now, i have changed the code as you mentioned in the second paragraph.     context.output(row) inside messages.forEach, after setting table row  as</p>
  
  <pre><code>List&lt;Message&gt; messages = feed.getMessage();
          messages.forEach((message) -&gt; {
              TableRow row = new TableRow();
              row.set("column2", message.getEventItemMap().get("key1"));
              context.output(row);
              }
  </code></pre>
  
  <p>Now, when i try to write this collection to BigQuery, as </p>
  
  <pre><code>rows.apply(BigQueryIO.writeTableRows().to(getTable(projectId, datasetId, tableName)).withSchema(getSchema())
                      .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                      .withWriteDisposition(WriteDisposition.WRITE_APPEND));
  </code></pre>
  
  <p>I am getting the below exception. </p>
  
  <pre><code>Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.NullPointerException
  at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:331)
  at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:301)
  at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:200)
  at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:63)
  at org.apache.beam.sdk.Pipeline.run(Pipeline.java:297)
  at org.apache.beam.sdk.Pipeline.run(Pipeline.java:283)
  at com.chefd.gcloud.analytics.pipeline.MyPipeline.main(MyPipeline.java:284)
  
  
  Caused by: java.lang.NullPointerException
  at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl.insertAll(BigQueryServicesImpl.java:759)
  at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl.insertAll(BigQueryServicesImpl.java:809)
  at org.apache.beam.sdk.io.gcp.bigquery.StreamingWriteFn.flushRows(StreamingWriteFn.java:126)
  at org.apache.beam.sdk.io.gcp.bigquery.StreamingWriteFn.finishBundle(StreamingWriteFn.java:96)
  </code></pre>
  
  <p>Please help.</p>
  
  <p>Thank you. </p>