error on _resolve_should_track_driver_status with spark standalone (cluster mode)

[rlebreto]

Apache Airflow Provider(s)

apache-spark

Versions of Apache Airflow Providers

apache-airflow-providers-apache-spark 5.0.0
pyspark 3.5.4

Apache Airflow version

2.10.3

Operating System

ubuntu server 22.04

Deployment

Other

Deployment details

NA

What happened

Hi

SparkSubmitOperator fails when interacting with a spark standalone server in deploy-mode=cluster.

In such condition, the _resolve_should_track_driver_status function in /usr/local/lib/python3.12/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py returns True...

I have exactly the same pb as the one discussed here :
#21799

I did find a case opened for this issue.

What you think should happen instead

I have an error when _resolve_should_track_driver_status returns True.
If I modify the function to return False, testing the task passes but in such case the final spark job status is not confirmed by the spark driver.

How to reproduce

cat sparkoperator_test.py
import airflow
from datetime import datetime, timedelta

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

default_args = {
    'owner': 'xxxxxxxx',
    'start_date': datetime(2025, 1, 31),
}

with airflow.DAG('sparkoperator_test',
                  default_args=default_args,
                  schedule_interval='@hourly',
                  description='test of sparkoperator in airflow by Omar and Regis',
                  ) as dag:

    spark_conf={
        'spark.standalone.submit.waitAppCompletion':'true',
    }

    spark_compute_pi = SparkSubmitOperator(
        task_id='spark-compute-pi',
        conn_id='spark_qa1',
        java_class='org.apache.spark.examples.SparkPi',
        application='/opt/spark/examples/jars/spark-examples_2.12-3.5.3.jar',
        name='compute_pi_from_airflow',
        application_args=["30"],
        total_executor_cores=4,
        executor_memory='1g',
        executor_cores=1,
        verbose='true',
        conf=spark_conf,
        execution_timeout=timedelta(minutes=10),
        retries=0,
    )

    spark_compute_pi

spark version : 3.5.3
1 single node (hosting the master + 1 worker node)

The error happens when testing the task with airflow tasks test ...

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[kgw7401]

i want to get assigned this issue

[rlebreto]

Hi @kgw7401

Thanks a lot for your help.

To not be blocked, I forced the value returned by _resolve_should_track_driver_status from /usr/local/lib/python3.12/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py to False

I'm not completely sure of the impact of the problem.
I'm suspecting that airflow will not be able to get back the status from the driver if the connection with it is lost.

I hope that running my sparkSubmit job with verbose='true' + retries=0 will allow me to track an error if the job fails, but I'm not very confident with that :-)

[meetic-mrobin]

@rlebreto and I found a solution that works, at least for our usecase (standalone, deploy_mode cluster, Spark 3.5).

Our conclusion is that spark-submit --master XXX --status YYY is broken (or we didn't understand how it's supposed to operate):

• if you provide a master with the binary port 7077, it fails because it tries to query it in HTTP
• if you provide a master with the HTTP port 6066, it fails because RestClient does not actually understand what HTTP Api returns

So we had to make sure Airflow tracks driver status using Curl strategy. Which means enters this condition:

if spark_host.endswith(":6066"):

https://github.com/apache/airflow/blob/main/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L486

So we have to set port to 6066 in our Spark connector (ie. REST Api port).

In order to have job submission that works with this port, we added the following config flag in our DAG:

spark_conf={
    'spark.master.rest.enabled':'true',
    ...
}

spark_compute_pi = SparkSubmitOperator(
    conf=spark_conf,
    ...
)

We also had to enable REST API on spark server side (it's disabled by default in recent versions).