sdks/python/apache_beam/io/gcp/big_query_query_to_table_pipeline.py
def run_bq_pipeline(argv=None):
  """Run the sample BigQuery pipeline.

  Args:
    argv: Arguments to the run function.
  """
  parser = argparse.ArgumentParser()
  parser.add_argument('--query', required=True,
                      help='Query to process for the table.')
  parser.add_argument('--output', required=True,
                      help='Output BQ table to write results to.')
  parser.add_argument('--output_schema', dest='output_schema', required=True,
                      help='Schema for output BQ table.')
  parser.add_argument('--use_standard_sql', action='store_true',
                      dest='use_standard_sql',
                      help='Output BQ table to write results to.')
  parser.add_argument('--kms_key', default=None,
                      help='Use this Cloud KMS key with BigQuery.')
  parser.add_argument('--bq_temp_location',
                      default=None,
                      help=('GCS bucket to use to store files for '
                            'loading data into BigQuery.'))
  known_args, pipeline_args = parser.parse_known_args(argv)

  table_schema = parse_table_schema_from_json(known_args.output_schema)
  kms_key = known_args.kms_key

  p = TestPipeline(options=PipelineOptions(pipeline_args))

  # pylint: disable=expression-not-assigned
  # pylint: disable=bad-continuation
  (p | 'read' >> beam.io.Read(beam.io.BigQuerySource(
      query=known_args.query, use_standard_sql=known_args.use_standard_sql,
      kms_key=kms_key))
   | 'write' >> beam.io.WriteToBigQuery(
           known_args.output,
           schema=table_schema,
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           write_disposition=beam.io.BigQueryDisposition.WRITE_EMPTY,
           gs_location=known_args.bq_temp_location))

  result = p.run()
  result.wait_until_finish()
Run the sample BigQuery pipeline.

  Args:
    argv: Arguments to the run function.
  
Run the sample BigQuery pipeline.
Run the sample BigQuery pipeline .
def run_bq_pipeline(argv=None):
  parser = argparse.ArgumentParser()
  parser.add_argument('--query', required=True,
                      help='Query to process for the table.')
  parser.add_argument('--output', required=True,
                      help='Output BQ table to write results to.')
  parser.add_argument('--output_schema', dest='output_schema', required=True,
                      help='Schema for output BQ table.')
  parser.add_argument('--use_standard_sql', action='store_true',
                      dest='use_standard_sql',
                      help='Output BQ table to write results to.')
  parser.add_argument('--kms_key', default=None,
                      help='Use this Cloud KMS key with BigQuery.')
  parser.add_argument('--bq_temp_location',
                      default=None,
                      help=('GCS bucket to use to store files for '
                            'loading data into BigQuery.'))
  known_args, pipeline_args = parser.parse_known_args(argv)
  table_schema = parse_table_schema_from_json(known_args.output_schema)
  kms_key = known_args.kms_key
  p = TestPipeline(options=PipelineOptions(pipeline_args))
  (p | 'read' >> beam.io.Read(beam.io.BigQuerySource(
      query=known_args.query, use_standard_sql=known_args.use_standard_sql,
      kms_key=kms_key))
   | 'write' >> beam.io.WriteToBigQuery(
           known_args.output,
           schema=table_schema,
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           write_disposition=beam.io.BigQueryDisposition.WRITE_EMPTY,
           gs_location=known_args.bq_temp_location))
  result = p.run()
  result.wait_until_finish()
def run_bq_pipeline ( argv = None ) : parser = argparse . ArgumentParser ( ) parser . add_argument ( '--query' , required = True , help = 'Query to process for the table.' ) parser . add_argument ( '--output' , required = True , help = 'Output BQ table to write results to.' ) parser . add_argument ( '--output_schema' , dest = 'output_schema' , required = True , help = 'Schema for output BQ table.' ) parser . add_argument ( '--use_standard_sql' , action = 'store_true' , dest = 'use_standard_sql' , help = 'Output BQ table to write results to.' ) parser . add_argument ( '--kms_key' , default = None , help = 'Use this Cloud KMS key with BigQuery.' ) parser . add_argument ( '--bq_temp_location' , default = None , help = ( 'GCS bucket to use to store files for ' 'loading data into BigQuery.' ) ) known_args , pipeline_args = parser . parse_known_args ( argv ) table_schema = parse_table_schema_from_json ( known_args . output_schema ) kms_key = known_args . kms_key p = TestPipeline ( options = PipelineOptions ( pipeline_args ) ) ( p | 'read' >> beam . io . Read ( beam . io . BigQuerySource ( query = known_args . query , use_standard_sql = known_args . use_standard_sql , kms_key = kms_key ) ) | 'write' >> beam . io . WriteToBigQuery ( known_args . output , schema = table_schema , create_disposition = beam . io . BigQueryDisposition . CREATE_IF_NEEDED , write_disposition = beam . io . BigQueryDisposition . WRITE_EMPTY , gs_location = known_args . bq_temp_location ) ) result = p . run ( ) result . wait_until_finish ( )
Run the sample BigQuery pipeline.
Run the sample BigQuery pipeline .
"""Run the sample BigQuery pipeline.

  Args:
    argv: Arguments to the run function.
  """ # pylint: disable=expression-not-assigned # pylint: disable=bad-continuation
argv
Arguments to the run function.
Arguments to the run function .