Start Dataflow Job

start_dataflow.py

def start_dataflow_pipeline(): 
  bucket = ‘fb_bucket’
  BODY = {
    “jobName”: ‘fb_
    catalog_upload’,
    “gcsPath”: “gs://{bucket}/catalog_up”.format(bucket=bucket),
    “environment”: {
      “tempLocation”: “gs://{bucket}/temp”.format(bucket=bucket),
      “zone”: “us-central1-f”
    }
  }
  credentials = GoogleCredentials.get_application_default()
  service = build(‘dataflow’, ‘v1b3’, credentials=credentials)
  dfrequest = service.projects().templates().create(projectId=settings.PROJECT_ID, body=BODY)
  dfresponse = dfrequest.execute()
  return dfresponse