Worker Sample for BigQuery Export

example_dataproc_twitter_worker_bq.py

import utils
from flask import Flask, request 
from config import config
from connector.gcp import GCPService
from scheduler import SchedulerJob


app = Flask(__name__)
gcp_service = GCPService() 
scheduler = SchedulerJob()


@app.route("/export_customers", methods=['POST'])
def export():
    """Runs a query against BigQuery and export results to a GCS bucket."""
    date = (None if request.form.get('date') == 'None' else
        utils.process_url_date(request.form.get('date')))
    query_job_body = utils.load_query_job_body(date,
        **config)
    job = gcp_service.bigquery.execute_job(config['general']['project_id'],
        query_job_body)

    gcp_service.bigquery.poll_job(job)

    extract_job_body = utils.load_extract_job_body(date, **config)
    gcp_service.bigquery.execute_job(config['general']['project_id'],
        extract_job_body)
    return "finished"