title: Apache Beam Pipeline to read from REST API and write to Google Big Query
body: <p>I've been trying to get my pipeline to run with a classic template on Dataflow.</p>
  <p>The pipeline is supposed to read a REST API. The answer coming back from the API should then be written into a bigquery table.</p>
  <p>It is running successfully when I limits the api to 1 result, however, if I increase the limit to more than 1, it return as error</p>
  <p>Help much appreciated - here is my pipeline code:</p>
  <pre><code>#!/usr/bin/env python3
  
  import apache_beam as beam
  #from apache_beam.options.pipeline_options import GoogleCloudOptions
  from apache_beam.options.pipeline_options import PipelineOptions
  from apache_beam.io.gcp.internal.clients import bigquery
  import setuptools
  
  import logging
  
  # Handling of API calls
  import json
  import requests
  
  class get_api_data(beam.DoFn):
      def __init__(self):
          logging.debug(&quot;fetching api data&quot;)
  
      def process(self,dummy_start):
  
          api_url = &quot;https://api.thedogapi.com/v1/breeds/?limit=05&quot;
  
          logging.debug(&quot;Now fetching from &quot;, api_url)
  
          response = requests.get(api_url)
  
          return list(response.json())
  
  
  
  def run(argv=None):
  
      beam_options = PipelineOptions(
          runner='direct', #'DataflowRunner',
          region='us-west2',
          project='projectid',
          job_name='api-to-bq-test1', # Dataflow job name
          temp_location='gs://rb-munish-playground/temp',
      )
  
      # BigQuery Table details
      table_spec = bigquery.TableReference(
      projectId='projectid',
      datasetId='dataflow_test',
      tableId='dogs')
  
      table_schema = {
      'fields': [
      {
          'name': 'weight', 'type': 'RECORD', 'mode': 'REPEATED',
              'fields': [{'name':'imperial','type':'STRING','mode':'NULLABLE'},
                          {'name':'metric','type':'STRING','mode':'NULLABLE'}]
      },
      {
          'name': 'height', 'type': 'RECORD', 'mode': 'REPEATED',
              'fields': [{'name':'imperial','type':'STRING','mode':'NULLABLE'},
                          {'name':'metric','type':'STRING','mode':'NULLABLE'}]
      },
      {
          'name': 'id', 'type': 'INTEGER', 'mode': 'REQUIRED'
      },
      {
          'name': 'name', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'bred_for', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'breed_group', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'life_span', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'temperament', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'origin', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'reference_image_id', 'type': 'STRING', 'mode': 'NULLABLE'
      },
      {
          'name': 'image', 'type': 'RECORD', 'mode': 'REPEATED',
              'fields': [{'name':'height','type':'INTEGER','mode':'NULLABLE'},
                          {'name':'id','type':'STRING','mode':'NULLABLE'},
                          {'name':'url','type':'STRING','mode':'NULLABLE'},
                          {'name':'width','type':'INTEGER','mode':'NULLABLE'}]
      },
      ]
  }
  
      # BigQueryDisposition.CREATE_IF_NEEDED:
      # Specifies that the write operation should create a new table if one does not exist.
      # If you use this value, you must provide a table schema.
      # CREATE_IF_NEEDED is the default behavior.
  
      # BigQueryDisposition.CREATE_NEVER:
      # Specifies that a table should never be created.
      # If the destination table does not exist, the write operation fails.
  
      # BigQueryDisposition.WRITE_EMPTY:
      # Specifies that the write operation should fail at runtime if the destination table is not empty.
      # WRITE_EMPTY is the default behavior.
  
      # BigQueryDisposition.WRITE_TRUNCATE:
      # Specifies that the write operation should replace an existing table.
      # Any existing rows in the destination table are removed, and the new rows are added to the table.
  
      # BigQueryDisposition.WRITE_APPEND:
      # Specifies that the write operation should append the rows to the end of the existing table.
  
      #p1 = beam.Pipeline(options=beam_options)
      with beam.Pipeline(options=beam_options) as pipeline:
  
          ingest_data = (
              pipeline
              | 'Create' &gt;&gt; beam.Create(['Start'])  # workaround to kickstart the pipeline
              | 'fetch API data' &gt;&gt; beam.ParDo(get_api_data())
              | 'write into gbq' &gt;&gt; beam.io.WriteToBigQuery(table = table_spec, schema=table_schema, write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE ,create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
              #| 'write to text' &gt;&gt; beam.io.WriteToText(&quot;./results.txt&quot;)
          )
  
          result = pipeline.run()
  
  if __name__ == '__main__':
      logging.getLogger().setLevel(logging.INFO)
      run()
  </code></pre>
  <p>Here is the error stack</p>
  <pre><code>  File &quot;/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery_file_loads.py&quot;, line 742, in process
      self.bq_wrapper.wait_for_bq_job(ref, sleep_duration_sec=10, max_retries=0)
    File &quot;/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery_tools.py&quot;, line 627, in wait_for_bq_job
      raise RuntimeError(
  RuntimeError: BigQuery job beam_bq_job_LOAD_apitobqtest1_LOAD_STEP_465_21cc6601da786b6d152c6a0338a36bf7_1d78c9d4d4674091a7952fd88df97035 failed. Error Result: &lt;ErrorProto
   location: 'gs://rb-munish-playground/temp/bq_load/843e8ffd21e346d5b754d3d15dff14da/rb-munish-playground.dataflow_test.dogs/35808f33-0892-4fe6-8cb0-5f90d7d11a89'
   message: 'Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 2; errors: 1. Please look into the errors[] collection for more details.'
   reason: 'invalid'&gt; [while running 'write into gbq/BigQueryBatchFileLoads/WaitForDestinationLoadJobs']
  </code></pre>