'Executes a BigQuery load command to load data from Google Cloud Storage
to BigQuery. See here:
https://cloud.google.com/bigquery/docs/reference/v2/jobs
For more details about these parameters.
:param destination_project_dataset_table:
The dotted (<project>.|<project>:)<dataset>.<table> BigQuery table to load
data into. If <project> is not included, project will be the project defined
in the connection json.
:type destination_project_dataset_table: string
:param schema_fields: The schema field list as defined here:
https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.load
:type schema_fields: list
:param source_uris: The source Google Cloud
Storage URI (e.g. gs://some-bucket/some-file.txt). A single wild
per-object name can be used.
:type source_uris: list
:param source_format: File format to export.
:type source_format: string
:param create_disposition: The create disposition if the table doesn\'t exist.
:type create_disposition: string
:param skip_leading_rows: Number of rows to skip when loading from a CSV.
:type skip_leading_rows: int
:param write_disposition: The write disposition if the table already exists.
:type write_disposition: string
:param field_delimiter: The delimiter to use when loading from a CSV.
:type field_delimiter: string
:param max_bad_records: The maximum number of bad records that BigQuery can
ignore when running the job.
:type max_bad_records: int
:param quote_character: The value that is used to quote data sections in a CSV file.
:type quote_character: string
:param schema_update_options: Allows the schema of the desitination
table to be updated as a side effect of the load job.
:type schema_update_options: list'

def run_load(self, destination_project_dataset_table, schema_fields, source_uris, source_format='CSV', create_disposition='CREATE_IF_NEEDED', skip_leading_rows=0, write_disposition='WRITE_EMPTY', field_delimiter=',', max_bad_records=0, quote_character=None, schema_update_options=()):

    source_format = source_format.upper()
    allowed_formats = ['CSV', 'NEWLINE_DELIMITED_JSON', 'AVRO', 'GOOGLE_SHEETS']
    if (source_format not in allowed_formats):
        raise ValueError('{0}    is    not    a    valid    source    format.    Please    use    one    of    the    following    types:    {1}'.format(source_format, allowed_formats))
    allowed_schema_update_options = ['ALLOW_FIELD_ADDITION', 'ALLOW_FIELD_RELAXATION']
    if (not set(allowed_schema_update_options).issuperset(set(schema_update_options))):
        raise ValueError('{0}    contains    invalid    schema    update    options.    Please    only    use    one    or    more    of    the    following    options:    {1}'.format(schema_update_options, allowed_schema_update_options))
    (destination_project, destination_dataset, destination_table) = _split_tablename(table_input=destination_project_dataset_table, default_project_id=self.project_id, var_name='destination_project_dataset_table')
    configuration = {'load': {'createDisposition': create_disposition, 'destinationTable': {'projectId': destination_project, 'datasetId': destination_dataset, 'tableId': destination_table}, 'sourceFormat': source_format, 'sourceUris': source_uris, 'writeDisposition': write_disposition}}
    if schema_fields:
        configuration['load']['schema'] = {'fields': schema_fields}
    if schema_update_options:
        if (write_disposition not in ['WRITE_APPEND', 'WRITE_TRUNCATE']):
            raise ValueError("schema_update_options    is    only    allowed    if    write_disposition    is    'WRITE_APPEND'    or    'WRITE_TRUNCATE'.")
        else:
            logging.info("Adding    experimental    'schemaUpdateOptions':    {0}".format(schema_update_options))
            configuration['load']['schemaUpdateOptions'] = schema_update_options
    if (source_format == 'CSV'):
        configuration['load']['skipLeadingRows'] = skip_leading_rows
        configuration['load']['fieldDelimiter'] = field_delimiter
    if max_bad_records:
        configuration['load']['maxBadRecords'] = max_bad_records
    if quote_character:
        configuration['load']['quote'] = quote_character
    return self.run_with_configuration(configuration)
