title: Writing to BigQuery from Cloud Dataflow: Unable to create a side-input view from input
body: <p>I'm trying to write a datastore flow that reads in a stream for pub sub and writes in into big query.</p>
  
  <p>When trying to run the tool I get the error " Unable to create a side-input view from input" with the stack trace:</p>
  
  <pre><code>Exception in thread "main" java.lang.IllegalStateException: Unable to create a side-input view from input
  at com.google.cloud.dataflow.sdk.transforms.View$AsIterable.validate(View.java:277)
  at com.google.cloud.dataflow.sdk.transforms.View$AsIterable.validate(View.java:268)
  at com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:366)
  at com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:274)
  at com.google.cloud.dataflow.sdk.values.PCollection.apply(PCollection.java:161)
  at com.google.cloud.dataflow.sdk.io.Write$Bound.createWrite(Write.java:214)
  at com.google.cloud.dataflow.sdk.io.Write$Bound.apply(Write.java:79)
  at com.google.cloud.dataflow.sdk.io.Write$Bound.apply(Write.java:68)
  at com.google.cloud.dataflow.sdk.runners.PipelineRunner.apply(PipelineRunner.java:74)
  at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.apply(DirectPipelineRunner.java:247)
  at com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:367)
  at com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:290)
  at com.google.cloud.dataflow.sdk.values.PCollection.apply(PCollection.java:174)
  at com.google.cloud.dataflow.sdk.io.BigQueryIO$Write$Bound.apply(BigQueryIO.java:1738)
  at com.google.cloud.dataflow.sdk.io.BigQueryIO$Write$Bound.apply(BigQueryIO.java:1440)
  at com.google.cloud.dataflow.sdk.runners.PipelineRunner.apply(PipelineRunner.java:74)
  at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.apply(DirectPipelineRunner.java:247)
  at com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:367)
  at com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:274)
  at com.google.cloud.dataflow.sdk.values.PCollection.apply(PCollection.java:161)
  at co.uk.bubblestudent.dataflow.StarterPipeline.main(StarterPipeline.java:116)
  Caused by: java.lang.IllegalStateException: GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.
  at com.google.cloud.dataflow.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:192)
  at com.google.cloud.dataflow.sdk.transforms.View$AsIterable.validate(View.java:275)
  ... 20 more
  </code></pre>
  
  <p>My code is:</p>
  
  <pre><code>public class StarterPipeline {
  
  
  public static final Duration ONE_DAY = Duration.standardDays(1);
  public static final Duration ONE_HOUR = Duration.standardHours(1);
  public static final Duration TEN_SECONDS = Duration.standardSeconds(10);
  private static final Logger LOG = LoggerFactory.getLogger(StarterPipeline.class);
  
    private static TableSchema schemaGen() {
      List&lt;TableFieldSchema&gt; fields = new ArrayList&lt;&gt;();
      fields.add(new TableFieldSchema().setName("facebookID").setType("STRING"));
      fields.add(new TableFieldSchema().setName("propertyID").setType("STRING"));
      fields.add(new TableFieldSchema().setName("time").setType("TIMESTAMP"));
      TableSchema schema = new TableSchema().setFields(fields);
      return schema;
    }
  
    public static void main(String[] args) {
    LOG.info("Starting");
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    LOG.info("Pipeline made");
    // For Cloud execution, set the Cloud Platform project, staging location,
    // and specify DataflowPipelineRunner or BlockingDataflowPipelineRunner.
    options.setProject(&lt;project&gt;);
    options.setStagingLocation(&lt;bucket&gt;);
    options.setTempLocation(&lt;bucket&gt;);
    Pipeline p = Pipeline.create(options);
  
  
    TableSchema schema = schemaGen();
    LOG.info("Schema made");
    try {
      LOG.info(schema.toPrettyString());
  } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
  }
    PCollection&lt;String&gt; input = p.apply(PubsubIO.Read.named("ReadFromPubsub").subscription(&lt;subscription&gt;));
  
    PCollection&lt;TableRow&gt; pardo = input.apply(ParDo.of(new FormatAsTableRowFn()));
    LOG.info("Formatted Row");
  
    pardo.apply(BigQueryIO.Write.named("Write into BigQuery").to(&lt;table&gt;)
         .withSchema(schema)
         .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
         .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
    LOG.info("about to run");
    p.run();
  
    }
  
  
    static class FormatAsTableRowFn extends DoFn&lt;String, TableRow&gt; {
      @Override
      public void processElement(ProcessContext c) {
          LOG.info("Formatting");
           String json = c.element();
  
           //HashMap&lt;String,String&gt; items = new Gson().fromJson(json, new TypeToken&lt;HashMap&lt;String, String&gt;&gt;(){}.getType());
  
          // Make a BigQuery row from the JSON object:
          TableRow row = new TableRow()
              .set("facebookID","324234")
              .set("properttyID", "23423")
              .set("time", "12312313123");
  
  
         /*
          *     TableRow row = new TableRow()
              .set("facebookID", items.get("facbookID"))
              .set("properttyID", items.get("propertyID"))
              .set("time", items.get("time"));
          */
          c.output(row);
      }
    }
  }
  </code></pre>
  
  <p>Any suggestions on what this might be?</p>