Create dataproc serverless spark batches operator #19248

MaksYermak · 2021-10-27T10:35:36Z

Create operator for working with Batches for Google Dataproc. Includes operators, hooks, example dags, tests and docs.

Co-authored-by: Wojciech Januszek januszek@google.com
Co-authored-by: Lukasz Wyszomirski wyszomirski@google.com
Co-authored-by: Maksim Yermakou maksimy@google.com

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

boring-cyborg · 2021-10-27T10:35:40Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

josh-fell · 2021-10-27T14:58:35Z

airflow/providers/google/cloud/operators/dataproc.py

+    :type timeout: float
+    :param metadata: Additional metadata that is provided to the method.
+    :type metadata: Sequence[Tuple[str, str]]
+    """


Missing gcp_conn_id and impersonation_chain from the docstring.

@josh-fell I've added it

airflow/providers/google/cloud/example_dags/example_dataproc.py

josh-fell · 2021-10-28T15:04:23Z

@MaksYermak General thoughts across the new operators:

WDYT about adding batch_id as a template_field? Would it be beneficial for users to be able to dynamically create batch_id values?
Adjacently, is batch_id something the DataprocCreateBatchOperator can push as an XCom such that the value can be used in downstream tasks as an XComArgs (i.e. the use case in the example DAG if the .output property is used as an input to the DataprocGetBatchOperator and DataprocDeleteBatchOperator tasks)?

Not saying it needs to/should be implemented but thought these might be good (and possibly cheap) features.

airflow/providers/google/cloud/operators/dataproc.py

MaksYermak · 2021-11-02T14:50:53Z

@MaksYermak General thoughts across the new operators:

WDYT about adding batch_id as a template_field? Would it be beneficial for users to be able to dynamically create batch_id values?

Adjacently, is batch_id something the DataprocCreateBatchOperator can push as an XCom such that the value can be used in downstream tasks as an XComArgs (i.e. the use case in the example DAG if the .output property is used as an input to the DataprocGetBatchOperator and DataprocDeleteBatchOperator tasks)?

Not saying it needs to/should be implemented but thought these might be good (and possibly cheap) features.

@josh-fell answers for your questions

I think it can be useful I will add it to a code.
No, because DataprocCreateBatchOperator returns a Batch object, not a batch_id

MaksYermak · 2021-11-15T10:19:25Z

@turbaszek @josh-fell @mik-laj hi guys, could you look on this PR one more time?

josh-fell · 2021-11-16T15:22:53Z

@MaksYermak Should batch_id be added to template_fields in DataprocGetBatchOperator and DataprocDeleteBatchOperator as well? No strong opinion though.

LGTM 👍

MaksYermak · 2021-11-17T09:29:39Z

@MaksYermak Should batch_id be added to template_fields in DataprocGetBatchOperator and DataprocDeleteBatchOperator as well? No strong opinion though.

LGTM 👍

@josh-fell make sense, I have added it to the code.

MaksYermak · 2021-11-19T07:44:23Z

@turbaszek @josh-fell @mik-laj hi guys, could you look and approve this PR for merge if all good?

lwyszomi · 2021-11-22T11:01:40Z

@turbaszek @josh-fell @mik-laj Is the a chance to merge this?

josh-fell · 2021-11-23T15:29:00Z

@MaksYermak @lwyszomi There is a static check that is failing. Can you address this?

FYI - I am not able to merge. A code owner will have to do that.

boring-cyborg · 2021-11-26T20:52:37Z

Awesome work, congrats on your first merged pull request!

(cherry picked from commit bf68b9a)

aoelvp94 · 2022-03-10T01:41:36Z

A question here, how can I specify the docker image to run pyspark workloads in custom containers using DataprocCreateBatchOperator?

lwyszomi · 2022-03-10T07:01:12Z

@aoelvp94 I'm not a expert of Dataproc serverless service, but operator is only a wrapper for the SDK and we take the same parameters as we heve there. So in the Batch definision we have runtimeConfig where you can specify the image. I hope this will help you.

Refs:
Batch -> https://cloud.google.com/dataproc-serverless/docs/reference/rest/v1/projects.locations.batches#Batch
runtimeConfig -> https://cloud.google.com/dataproc-serverless/docs/reference/rest/v1/projects.locations.batches#RuntimeConfig

aoelvp94 · 2022-03-10T12:41:19Z

Yeah @lwyszomi I am trying with something like that. I am trying that now so if I have news I will leave the code snippet for future references

aoelvp94 · 2022-03-11T05:15:21Z

@lwyszomi No success when I tried to set jinja params in the Batch object, do you know why?

lwyszomi · 2022-03-14T07:09:04Z

@aoelvp94 which version of the provider you are using? The Batch object is templated field starting from 6.4.0

aoelvp94 · 2022-03-14T11:15:45Z

@aoelvp94 which version of the provider you are using? The Batch object is templated field starting from 6.4.0

I am using composer-2.0.5-airflow-2.2.3 so I have 6.4.0

aoelvp94 · 2022-03-14T11:18:26Z

Another question, why the operator don't generate a hash in batch_id field as happens in DataprocSubmitPySparkJobOperator (I am replacing this implementation)? I have to delete the batch to run again the same batch (I prefer having the historical runs)

lwyszomi · 2022-03-14T11:39:36Z

@aoelvp94 which version of the provider you are using? The Batch object is templated field starting from 6.4.0

I am using composer-2.0.5-airflow-2.2.3 so I have 6.4.0

I will need more investigation why jijna paramams doesn't work, I checked and you have right that for Composer 2.0.5 using 6.4.0.

Another question, why the operator don't generate a hash in batch_id field as happens in DataprocSubmitPySparkJobOperator (I am replacing this implementation)? I have to delete the batch to run again the same batch (I prefer having the historical runs)

We created operators based on the exisiting SDK, we didn't add any extra logic to add hash to the batch_id.

lwyszomi · 2022-03-17T09:37:35Z

@aoelvp94 for which property you want to use jinja template?

aoelvp94 · 2022-03-17T09:42:03Z

batch when I use Batch / PySparkBatch object

lwyszomi · 2022-03-17T09:43:56Z

@aoelvp94 butch is tepletized so it should work, but which property inside Batch/PySparkBatch you try to templetize

lwyszomi · 2022-03-17T09:44:20Z

can you share Batch config?

aoelvp94 · 2022-03-17T10:27:21Z

I removed the main python file on purpose.

def get_week_ago(dt):
    bd = dt + relativedelta(days=-7)
    return bd.strftime('%Y-%m-%d')

dag = DAG(
        dag_id=f"{table_name}_table_v4",
        ...
        user_defined_macros={'week_ago': get_week_ago}
    )


create_batch = DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=project_id,
        region=f"{{{{ var.value.region_{environment} }}}}",
        batch=Batch(pyspark_batch=PySparkBatch(
        main_python_file_uri=main_python_file_uri,
        python_file_uris=files,
        args=[
            "--start-date",
            "{{ week_ago(execution_date) }}", # it doesn't work
            "--end-date",
            "{{ ds }}", # it doesn't work
            "--table",
            table_name,
            "--source-bucket",
            source_bucket,
            "--source-prefix",
            source_prefix,
            "--output-path",
            f"{output_path}/{table_name}/",
        ],
    ), runtime_config=RuntimeConfig(container_image="gcr.io/company/image_name:1.0.1")),
        batch_id=f"pyspark-table-{table_name.replace('_','-')}-{{{{ ds_nodash }}}}",
        retry=DEFAULT,
        timeout=500,
    )

lwyszomi · 2022-03-17T10:32:17Z

@aoelvp94 thanks, we will check why this not work, I will back to you with update when I will have any information

marekw-openx · 2022-03-17T14:01:47Z

@aoelvp94, have you tried passing batch config in the form of python dict? in my case I'm using just a python dict and it gets templetized, like:

batch = {
  "spark_batch": {
    "spark_job": {
      "args": ["-d", "{{ ts }}"]
    }
  }
  (...)
}

also, please double check the apache-airflow-providers-google version (batch wasn't templated in the 6.2.0)

MaksYermak · 2022-03-17T14:44:19Z

@aoelvp94 I have checked your configuration for correct work you should use dictionary instead of Batch() object. It is because in some reasons Airflow can't template object's property.

batch = {
    "pyspark_batch": {
        "main_python_file_uri": main_python_file_uri,
        "python_file_uris": files,
        "args": [
            "--start-date={{ get_week_ago(data_interval_start) }}",
            "--end-date={{ ds }}",
            (. . .)
        ],
    },
    ( . . .)
}

One more thing in the last Jinja version execution_date from the template is deprecated. Please use data_interval_start or logical_date instead.

MaksYermak requested review from mik-laj, turbaszek and vikramkoka as code owners October 27, 2021 10:35

boring-cyborg bot added area:providers kind:documentation provider:google Google (including GCP) related issues labels Oct 27, 2021

josh-fell reviewed Oct 28, 2021

View reviewed changes

turbaszek reviewed Oct 29, 2021

View reviewed changes

airflow/providers/google/cloud/operators/dataproc.py Show resolved Hide resolved

turbaszek reviewed Oct 29, 2021

View reviewed changes

airflow/providers/google/cloud/operators/dataproc.py Show resolved Hide resolved

MaksYermak force-pushed the create-dataproc-serverless-spark-batches-operator branch 2 times, most recently from 7d1ea91 to e9a956d Compare November 10, 2021 17:05

MaksYermak force-pushed the create-dataproc-serverless-spark-batches-operator branch from e9a956d to 4065232 Compare November 16, 2021 13:39

MaksYermak force-pushed the create-dataproc-serverless-spark-batches-operator branch from 4065232 to 0c1b5e5 Compare November 17, 2021 09:27

MaksYermak added 8 commits November 24, 2021 09:06

Create DataprocBatch hooks

8c094ef

Create DataprocBatch operators

28cda80

Fix code style

dee6f41

Change wait_for_operation function for Dataproc hooks

7b4c126

Update google-cloud-dataproc version

db21275

Add small fixes

60eafbf

Add 'batch_id' to template fields

6b78b8a

Change DataprocCreateBatchOperator logic

7b338ad

potiuk merged commit bf68b9a into apache:main Nov 26, 2021

potiuk mentioned this pull request Nov 30, 2021

Status of testing Providers that were prepared on November 30, 2021 #19895

Closed

44 tasks

dillonjohnson pushed a commit to dillonjohnson/airflow that referenced this pull request Dec 1, 2021

Create dataproc serverless spark batches operator (apache#19248)

42572f9

potiuk added the changelog:skip Changes that should be skipped from the changelog (CI, tests, etc..) label Jan 22, 2022

potiuk added this to the Airflow 2.2.4 milestone Jan 22, 2022

potiuk pushed a commit that referenced this pull request Jan 22, 2022

Create dataproc serverless spark batches operator (#19248)

a075bb7

(cherry picked from commit bf68b9a)

jedcunningham pushed a commit that referenced this pull request Jan 27, 2022

Create dataproc serverless spark batches operator (#19248)

67fc1d7

(cherry picked from commit bf68b9a)

aoelvp94 mentioned this pull request Mar 14, 2022

Google cloud dataproc: Add an operator to run a serverless spark in dataproc #22019

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create dataproc serverless spark batches operator #19248

Create dataproc serverless spark batches operator #19248

MaksYermak commented Oct 27, 2021 •

edited

Loading

boring-cyborg bot commented Oct 27, 2021

josh-fell Oct 27, 2021

MaksYermak Oct 29, 2021

josh-fell commented Oct 28, 2021 •

edited

Loading

MaksYermak commented Nov 2, 2021 •

edited

Loading

MaksYermak commented Nov 15, 2021

josh-fell commented Nov 16, 2021

MaksYermak commented Nov 17, 2021

MaksYermak commented Nov 19, 2021

lwyszomi commented Nov 22, 2021

josh-fell commented Nov 23, 2021

boring-cyborg bot commented Nov 26, 2021

aoelvp94 commented Mar 10, 2022

lwyszomi commented Mar 10, 2022

aoelvp94 commented Mar 10, 2022

aoelvp94 commented Mar 11, 2022 •

edited

Loading

lwyszomi commented Mar 14, 2022

aoelvp94 commented Mar 14, 2022

aoelvp94 commented Mar 14, 2022

lwyszomi commented Mar 14, 2022

lwyszomi commented Mar 17, 2022

aoelvp94 commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

aoelvp94 commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

marekw-openx commented Mar 17, 2022

MaksYermak commented Mar 17, 2022

Create dataproc serverless spark batches operator #19248

Create dataproc serverless spark batches operator #19248

Conversation

MaksYermak commented Oct 27, 2021 • edited Loading

boring-cyborg bot commented Oct 27, 2021

josh-fell Oct 27, 2021

Choose a reason for hiding this comment

MaksYermak Oct 29, 2021

Choose a reason for hiding this comment

josh-fell commented Oct 28, 2021 • edited Loading

MaksYermak commented Nov 2, 2021 • edited Loading

MaksYermak commented Nov 15, 2021

josh-fell commented Nov 16, 2021

MaksYermak commented Nov 17, 2021

MaksYermak commented Nov 19, 2021

lwyszomi commented Nov 22, 2021

josh-fell commented Nov 23, 2021

boring-cyborg bot commented Nov 26, 2021

aoelvp94 commented Mar 10, 2022

lwyszomi commented Mar 10, 2022

aoelvp94 commented Mar 10, 2022

aoelvp94 commented Mar 11, 2022 • edited Loading

lwyszomi commented Mar 14, 2022

aoelvp94 commented Mar 14, 2022

aoelvp94 commented Mar 14, 2022

lwyszomi commented Mar 14, 2022

lwyszomi commented Mar 17, 2022

aoelvp94 commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

aoelvp94 commented Mar 17, 2022

lwyszomi commented Mar 17, 2022

marekw-openx commented Mar 17, 2022

MaksYermak commented Mar 17, 2022

MaksYermak commented Oct 27, 2021 •

edited

Loading

josh-fell commented Oct 28, 2021 •

edited

Loading

MaksYermak commented Nov 2, 2021 •

edited

Loading

aoelvp94 commented Mar 11, 2022 •

edited

Loading