Database Reference
In-Depth Information
'tableId': source_table_id}
extract_config = {
'sourceTable': source_table_ref,
'destinationFormat': 'NEWLINE_DELIMITED_JSON',
'destinationUris': destination_uris}
return {'extract': extract_config}
def run_partitioned_extract_job(job_runner,
gcs_readers,
source_project_id, source_dataset_id,
source_table_id):
'''Runs a BigQuery extract job and reads the
results.'''
destination_uris = []
gcs_objects = []
timestamp = int(time.time())
partition_readers = []
for index in range(len(gcs_readers)):
gcs_object = 'output/%s.%s_%d.%d.*.json' % (
source_dataset_id,
source_table_id,
timestamp,
index)
gcs_objects.append(gcs_object)
destination_uris.append(gcs_readers[index].make_uri(gcs_object))
# Create the reader thread for this partition.
partition_readers.append(
PartitionReader(job_runner=job_runner,
gcs_reader=gcs_readers[index],
partition_id=index))
job_config = make_extract_config(source_project_id,
source_dataset_id,
source_table_id, destination_uris)
if not job_runner.start_job(job_config):
return
