'Create a new DataFlowPythonOperator. Note that both
dataflow_default_options and options will be merged to specify pipeline
execution parameter, and dataflow_default_options is expected to save
high-level options, for instances, project and zone information, which
apply to all dataflow operators in the DAG.
For more detail on job submission have a look at the reference:
https://cloud.google.com/dataflow/pipelines/specifying-exec-params
:param py_file: Reference to the python dataflow pipleline file, e.g.,
/some/local/file/path/to/your/python/pipeline/file.py.
:type py_file: string
:param py_options: Additional python options.
:type pyt_options: list of strings, e.g., ["-m", "-v"].
:param dataflow_default_options: Map of default job options.
:type dataflow_default_options: dict
:param options: Map of job specific options.
:type options: dict
:param gcp_conn_id: The connection ID to use connecting to Google Cloud
Platform.
:type gcp_conn_id: string
:param delegate_to: The account to impersonate, if any.
For this to work, the service account making the request must have
domain-wide  delegation enabled.
:type delegate_to: string'

@apply_defaults
def __init__(self, py_file, py_options=None, dataflow_default_options=None, options=None, gcp_conn_id='google_cloud_default', delegate_to=None, *args, **kwargs):

    super(DataFlowPythonOperator, self).__init__(*args, **kwargs)
    self.py_file = py_file
    self.py_options = (py_options or [])
    self.dataflow_default_options = (dataflow_default_options or {})
    self.options = (options or {})
    self.gcp_conn_id = gcp_conn_id
    self.delegate_to = delegate_to
