title: Google Dataflow write multiple line in BigQuery
body: <p>I have a simple flow which aim is to write two lines in one BigQuery Table.
  I use a DynamicDestinations because after that I will write on mutliple Table, on that example it's the same table...
  The problem is that I only have 1 line in my BigQuery table at the end.
  It stacktrace I see the following error on the second insert</p>
  
  <p>"
  status: {
     code:  6<br>
     message:  "Already Exists: Job sampleprojet3:b9912b9b05794aec8f4292b2ae493612_eeb0082ade6f4a58a14753d1cc92ddbc_00001-0"<br>
    }
  "</p>
  
  <p>What does it means ? 
  Is it related to this limitation ?
  <a href="https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/550" rel="nofollow noreferrer">https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/550</a>
  How can I do the job ?</p>
  
  <p>I use BeamSDK 2.0.0, I have try with 2.1.0 (same result)</p>
  
  <p>The way I launch : </p>
  
  <p>mvn compile exec:java -Dexec.mainClass=fr.gireve.dataflow.LogsFlowBug -Dexec.args="--runner=DataflowRunner --inputDir=gs://sampleprojet3.appspot.com/ --project=sampleprojet3 --stagingLocation=gs://dataflow-sampleprojet3/tmp" -Pdataflow-runner</p>
  
  <pre><code>    Pipeline p = Pipeline.create(options);
  
      final List&lt;String&gt; tableNameTableValue = Arrays.asList("table1:value1", "table1:value2", "table2:value1", "table2:value2");
  
      p.apply(Create.of(tableNameTableValue)).setCoder(StringUtf8Coder.of())
              .apply(BigQueryIO.&lt;String&gt;write()
              .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
              .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
              .to(new DynamicDestinations&lt;String, KV&lt;String, String&gt;&gt;() {
                  @Override
                  public KV&lt;String, String&gt; getDestination(ValueInSingleWindow&lt;String&gt; element) {
                      final String[] split = element.getValue().split(":");
                      return KV.of(split[0], split[1]) ;
                  }
  
                  @Override
                  public Coder&lt;KV&lt;String, String&gt;&gt; getDestinationCoder() {
                      return KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());
                  }
  
                  @Override
                  public TableDestination getTable(KV&lt;String, String&gt; row) {
                      String tableName = row.getKey();
                      String tableSpec = "sampleprojet3:testLoadJSON." + tableName;
                      return new TableDestination(tableSpec, "Table " + tableName);
                  }
  
                  @Override
                  public TableSchema getSchema(KV&lt;String, String&gt; row) {
                      List&lt;TableFieldSchema&gt; fields = new ArrayList&lt;&gt;();
                      fields.add(new TableFieldSchema().setName("myColumn").setType("STRING"));
                      TableSchema ts = new TableSchema();
                      ts.setFields(fields);
                      return ts;
                  }
              })
              .withFormatFunction(new SerializableFunction&lt;String, TableRow&gt;() {
                  public TableRow apply(String row) {
                      TableRow tr = new TableRow();
                      tr.set("myColumn", row);
                      return tr;
                  }
              }));
  
      p.run().waitUntilFinish();
  </code></pre>
  
  <p>Thanks</p>