Adding Support for Yarn queue and other extras in SparkSubmit Operator and Hook

[pateash]

Description

Spark-submit
--queue thequeue option specifies the YARN queue to which the application should be submitted.

more - https://spark.apache.org/docs/3.2.0/running-on-yarn.html

Use case/motivation

The --queue option is particularly useful in a multi-tenant environment where different users or groups have allocated resources in specific YARN queues.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[pateash]

Please assign this to me, this is a quick fix
I will try to complete this ASAP.

[hussein-awala]

@pateash assigned you.

[pateash]

Currently it's possible to set this option inside Spark connection, but we should have the ability to override it if needed using operator arguments.

[eladkal]

Currently it's possible to set this option inside Spark connection, but we should have the ability to override it if needed using operator arguments.

So do we have a task to complete here?

[pateash]

@eladkal,
I am thinking of adding it as an argument in SparkSubmitOperator if user wants to override it from connection extras.
let me know what you think.

[eladkal]

We normally allow to override connection extra from the operator level. For example with Snowflake you can set different warehouse than the default one set in the connection.

[pateash]

Yes, it's done using args AFAIK,
Let me add an argument to override this.

[databius]

I think the update caused the new issue since SparkSubmitOperator accidentally override the Base Operator's parameter.

https://github.com/apache/airflow/blob/main/airflow/providers/apache/spark/operators/spark_submit.py#L76

:param queue: The name of the YARN queue to which the application is submitted.
(will overwrite any yarn queue defined in the connection's extra JSON)

https://github.com/apache/airflow/blob/main/airflow/models/baseoperator.py#L587

:param queue: which queue to target when running this job. Not
 all executors implement queue management, the CeleryExecutor
 does support targeting specific queues.

So if you are using CeleryKubernetesExecutor, you cannot decide the Operator run on KubernetesExecutor by setting the queue to kubernetes_queue

[djuarezg]

Since 4.6.0 where this was introduced, onwards, using SparkSubmitOperator is broken for us, as the queue can no longer be taken into account due to this change.

Before, this queue parameter overrode BaseOperator value where a queue is an Airflow queue