title: How to use 'flatten' correctly in Dataflow
body: <p>Our pipeline looks like this:</p>
  
  <p><strong>GCS (gz compressed files) -> ParDo -> BigQuery</strong></p>
  
  <p>I want to use a 'flatten' to scoop in multiple files from GCS as the input to my pipeline. But it keeps borking with the error:</p>
  
  <pre><code>Workflow failed. Causes: (5001e5764f46ac2c): BigQuery creation of import job for table "Impressions_05_2015_denormalized_test" in dataset "CPT_XXXX" in project "gdfp-XXXX" failed. Causes: (5001e5764f46a1cf): Error:
   Message: Load configuration must specify at least one source URI
   HTTP Code: 400
  </code></pre>
  
  <p>Code:</p>
  
  <pre><code>PCollection&lt;String&gt; file1 = pipeline.apply(TextIO.Read.from("gs://&lt;bucket_name_removed&gt;/NetworkActiveViews_232503_20140918_21.gz").withCompressionType(TextIO.CompressionType.GZIP));
          PCollection&lt;String&gt; file2 = pipeline.apply(TextIO.Read.from("gs://&lt;bucket_name_removed&gt;/NetworkActiveViews_232503_20140918_22.gz").withCompressionType(TextIO.CompressionType.GZIP));
          PCollectionList&lt;String&gt; allFiles = PCollectionList.of(file1).and(file2);
          PCollection&lt;String&gt; inputRead = allFiles.apply(Flatten.&lt;String&gt;pCollections());
  inputRead.apply(ParDo.of(transformation)
                  .named(String.format("%s-CPT-transform", type))
                  .withSideInputs(views))
                  .apply(Write.to(getOutputTable(type))
                          .withCreateDisposition(CREATE_IF_NEEDED)
                          .withWriteDisposition(WRITE_APPEND)
                          .withSchema(schema)
                          .named(String.format("%s-BQ-write", type)));
  </code></pre>
  
  <p>Example job id: <strong>2015-05-12_19_54_06-10158770219525037626</strong></p>
  
  <p>What am I doing wrong?</p>