title: How to run multiple WriteToBigQuery parallel in google cloud dataflow / apache beam?
body: <p>I want to separate event from a bunch of multiple events, given data </p>
  
  <pre><code>{"type": "A", "k1": "v1"}
  {"type": "B", "k2": "v2"}
  {"type": "C", "k3": "v3"}
  </code></pre>
  
  <p>And I want to separate <code>type: A</code> events to table <code>A</code> in bigquery, <code>type:B</code> events to table <code>B</code>,  <code>type: C</code> events to table <code>C</code>.</p>
  
  <p>Here are my codes implemented through <code>apache beam</code> python sdk and write data to <code>bigquery</code>,</p>
  
  <pre><code>A_schema = 'type:string, k1:string'
  B_schema = 'type:string, k2:string'
  C_schema = 'type:string, k2:string'
  
  class ParseJsonDoFn(beam.DoFn):
      A_TYPE = 'tag_A'
      B_TYPE = 'tag_B'
      C_TYPE = 'tag_C'
      def process(self, element):
          text_line = element.trip()
          data = json.loads(text_line)
  
          if data['type'] == 'A':
              yield pvalue.TaggedOutput(self.A_TYPE, data)
          elif data['type'] == 'B':
              yield pvalue.TaggedOutput(self.B_TYPE, data)
          elif data['type'] == 'C':
              yield pvalue.TaggedOutput(self.C_TYPE, data)
  
  def run():
      parser = argparse.ArgumentParser()
      parser.add_argument('--input',
                        dest='input',
                        default='data/path/data',
                        help='Input file to process.')
      known_args, pipeline_args = parser.parse_known_args(argv)
      pipeline_args.extend([
        '--runner=DirectRunner',
        '--project=project-id',
        '--job_name=seperate-bi-events-job',
      ])
      pipeline_options = PipelineOptions(pipeline_args)
      pipeline_options.view_as(SetupOptions).save_main_session = True
      with beam.Pipeline(options=pipeline_options) as p:
          lines = p | ReadFromText(known_args.input)
  
      multiple_lines = (
          lines
          | 'ParseJSON' &gt;&gt; (beam.ParDo(ParseJsonDoFn()).with_outputs(
                                        ParseJsonDoFn.A_TYPE,
                                        ParseJsonDoFn.B_TYPE,
                                        ParseJsonDoFn.C_TYPE)))
  
      a_line = multiple_lines.tag_A
      b_line = multiple_lines.tag_B
      c_line = multiple_lines.tag_C
  
      (a_line
          | "output_a" &gt;&gt; beam.io.WriteToBigQuery(
                                            'temp.a',
                                            schema = A_schema,
                                            write_disposition = beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                                            create_disposition = beam.io.BigQueryDisposition.CREATE_IF_NEEDED
                                          ))
  
      (b_line
          | "output_b" &gt;&gt; beam.io.WriteToBigQuery(
                                            'temp.b',
                                            schema = B_schema,
                                            write_disposition = beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                                            create_disposition = beam.io.BigQueryDisposition.CREATE_IF_NEEDED
                                          ))
  
      (c_line
          | "output_c" &gt;&gt; beam.io.WriteToBigQuery(
                                            'temp.c',
                                            schema = (C_schema),
                                            write_disposition = beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                                            create_disposition = beam.io.BigQueryDisposition.CREATE_IF_NEEDED
                                          ))
  
      p.run().wait_until_finish()
  </code></pre>
  
  <p>The output:</p>
  
  <pre><code>INFO:root:start &lt;DoOperation output_banner/WriteToBigQuery output_tags=['out']&gt;
  INFO:oauth2client.transport:Attempting refresh to obtain initial access_token
  INFO:oauth2client.client:Refreshing access_token
  WARNING:root:Sleeping for 150 seconds before the write as BigQuery inserts can be routed to deleted table for 2 mins after the delete and create.
  INFO:root:start &lt;DoOperation output_banner/WriteToBigQuery output_tags=['out']&gt;
  INFO:oauth2client.transport:Attempting refresh to obtain initial access_token
  INFO:oauth2client.client:Refreshing access_token
  WARNING:root:Sleeping for 150 seconds before the write as BigQuery inserts can be routed to deleted table for 2 mins after the delete and create.
  INFO:root:start &lt;DoOperation output_banner/WriteToBigQuery output_tags=['out']&gt;
  INFO:oauth2client.transport:Attempting refresh to obtain initial access_token
  INFO:oauth2client.client:Refreshing access_token
  WARNING:root:Sleeping for 150 seconds before the write as BigQuery inserts can be routed to deleted table for 2 mins after the delete and create.
  </code></pre>
  
  <p>However, there are two issues here </p>
  
  <ul>
  <li>there is no data in <code>bigquery</code>?</li>
  <li>From the logs it seems the codes does NOT run parallel rather than run 3  times sequence?</li>
  </ul>
  
  <p>Is there something wrong in my codes or something am I missing?</p>