StreamingWriteFn: check if table exists before creating

In StreamingWriteFn, every worker tries to create the table
without checking beforehand whether the table exists. This
behavior can lead to temporarily violating the BigQuery API
quota limits on table.insert:

{
  "code" : 403,
  "errors" : [ {
    "domain" : "global",
    "location" : "table.write",
    "locationType" : "other",
    "message" : "Exceeded rate limits: Your table exceeded quota for table.insert or table.update per table. For more information, see https://cloud.google.com/bigquery/troubleshooting-errors";,
    "reason" : "rateLimitExceeded"
  } ],
  "message" : "Exceeded rate limits: Your table exceeded quota for table.insert or table.update per table. For more information, see https://cloud.google.com/bigquery/troubleshooting-errors";
}

Note that although this error appears severe, it should not cause
jobs to fail. This change primarily aims to reduce the occurrence of
this alarming log.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=112818882

Author: dhalperi

sdk/src/main/java/com/google/cloud/dataflow/sdk/io/BigQueryIO.java

@@ -31,6 +31,8 @@
 import com.google.cloud.dataflow.sdk.coders.TableRowJsonCoder;
 import com.google.cloud.dataflow.sdk.coders.VarIntCoder;
 import com.google.cloud.dataflow.sdk.coders.VoidCoder;
+import com.google.cloud.dataflow.sdk.io.BigQueryIO.Write.CreateDisposition;
+import com.google.cloud.dataflow.sdk.io.BigQueryIO.Write.WriteDisposition;
 import com.google.cloud.dataflow.sdk.options.BigQueryOptions;
 import com.google.cloud.dataflow.sdk.options.GcpOptions;
 import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
@@ -1086,7 +1088,8 @@ public TableReference getOrCreateTable(BigQueryOptions options, String tableSpec
             TableSchema tableSchema = JSON_FACTORY.fromString(jsonTableSchema, TableSchema.class);
             Bigquery client = Transport.newBigQueryClient(options).build();
             BigQueryTableInserter inserter = new BigQueryTableInserter(client);
-            inserter.tryCreateTable(tableReference, tableSchema);
+            inserter.getOrCreateTable(tableReference, WriteDisposition.WRITE_APPEND,
+                CreateDisposition.CREATE_IF_NEEDED, tableSchema);
             createdTables.add(tableSpec);
           }
         }