Database Reference
In-Depth Information
'''Run a bigquery job and update pipeline status.'''
global g_state
runner = JobRunner(PROJECT_ID,
job_id_prefix + '_' + job_type,
client=bigquery)
runner.start_job({job_type: config})
with g_state_lock:
g_state[job_type + '_job_id'] = runner.job_id
job_state = 'STARTED'
while job_state != 'DONE':
time.sleep(5)
result = runner.get_job()
job_state = result['status']['state']
with g_state_lock:
g_state[job_type + '_result'] =
pre(json.dumps(result, indent=2))
if 'errorResult' in result['status']:
raise
RuntimeError(json.dumps(result['status']['errorResult'],
indent=2))
def wait_for_pipeline(pipeline_id):
'''Wait for a MapReduce pipeline to complete.'''
mapreduce_id = None
while True:
time.sleep(5)
pipeline = MapperPipeline.from_id(pipeline_id)
if not mapreduce_id and
pipeline.outputs.job_id.filled:
mapreduce_id = pipeline.outputs.job_id.value
with g_state_lock:
g_state['mapper_link'] = (
'<a href="/mapreduce/
detail?mapreduce_id=%s">%s</a>' % (
mapreduce_id, mapreduce_id))
if pipeline.has_finalized:
break
