Spark-submit task fails on deploy mode cluster #21799

vmferreira81 · 2022-02-24T17:40:02Z

vmferreira81
Feb 24, 2022

Airflow v2.2.4 running on Linux (same error with Airflow v2.2.3)
Spark v3.1.2 running on Linux

Hi,

I'm trying to run a spark-submit task on a spark cluster. My Spark connection is configured with extra {"deploy-mode": "cluster"}.

My task is executed successfully on Spark, but when it finishes, I receive the following error:

[2022-02-24, 08:33:11 ] {spark_submit.py:488} INFO - 22/02/24 11:33:11 INFO ClientEndpoint: Driver successfully submitted as driver-20220224113311-0599
[2022-02-24, 08:33:16 ] {spark_submit.py:488} INFO - 22/02/24 11:33:16 INFO ClientEndpoint: State of driver-20220224113311-0599 is FINISHED
[2022-02-24, 08:33:16 ] {spark_submit.py:488} INFO - 22/02/24 11:33:16 INFO ClientEndpoint: State of driver driver-20220224113311-0599 is FINISHED, exiting spark-submit JVM.
[2022-02-24, 08:33:16 ] {spark_submit.py:488} INFO - 22/02/24 11:33:16 INFO ShutdownHookManager: Shutdown hook called
[2022-02-24, 08:33:16 ] {spark_submit.py:488} INFO - 22/02/24 11:33:16 INFO ShutdownHookManager: Deleting directory /tmp/spark-08e72231-8b3d-4842-9b86-09e85226aa64
[2022-02-24, 08:33:52 ] {taskinstance.py:1718} ERROR - Task failed with exception
Traceback (most recent call last):
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1334, in _run_raw_task
self._execute_task_with_callbacks(context)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1460, in _execute_task_with_callbacks
result = self._execute_task(context, self.task)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1516, in _execute_task
result = execute_callable(context=context)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/operators/spark_submit.py", line 157, in execute
self._hook.submit(self._application)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 436, in submit
self._start_driver_status_tracking()
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 575, in _start_driver_status_tracking
raise AirflowException(
airflow.exceptions.AirflowException: Failed to poll for the driver status 10 times: returncode = 1
[2022-02-24, 08:33:52 ] {taskinstance.py:1272} INFO - Marking task as FAILED. dag_id=teste_spark_submit_airflow, task_id=spark_processing, execution_date=20220224T143307, start_date=20220224T143308, end_date=20220224T143352
[2022-02-24, 08:33:52 ] {standard_task_runner.py:89} ERROR - Failed to execute job 9 for task spark_processing
Traceback (most recent call last):
File "/root/anaconda3/lib/python3.9/site-packages/airflow/task/task_runner/standard_task_runner.py", line 85, in _start_by_fork
args.func(args, dag=self.dag)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 48, in command
return func(*args, **kwargs)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/utils/cli.py", line 92, in wrapper
return f(*args, **kwargs)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 298, in task_run
_run_task_by_selected_method(args, dag, ti)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
_run_raw_task(args, ti)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 180, in _run_raw_task
ti._run_raw_task(
File "/root/anaconda3/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
return func(*args, session=session, **kwargs)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1334, in _run_raw_task
self._execute_task_with_callbacks(context)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1460, in _execute_task_with_callbacks
result = self._execute_task(context, self.task)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1516, in _execute_task
result = execute_callable(context=context)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/operators/spark_submit.py", line 157, in execute
self._hook.submit(self._application)
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 436, in submit
self._start_driver_status_tracking()
File "/root/anaconda3/lib/python3.9/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 575, in _start_driver_status_tracking
raise AirflowException(
airflow.exceptions.AirflowException: Failed to poll for the driver status 10 times: returncode = 1
[2022-02-24, 08:33:53 ] {local_task_job.py:154} INFO - Task exited with return code 1
[2022-02-24, 08:33:53 ] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check

When I remove the extra "deploy-mode: cluster", this error doesn't occur.

I saw that this in spark_submit.py code:

def _resolve_should_track_driver_status(self) -> bool: """ Determines whether or not this hook should poll the spark driver status through subsequent spark-submit status requests after the initial spark-submit request :return: if the driver status should be tracked """ return 'spark://' in self._connection['master'] and self._connection['deploy_mode'] == 'cluster'

How can I solve this? I really need to run on cluster mode.

Thank you

potiuk · 2022-02-27T18:34:34Z

potiuk
Feb 27, 2022
Collaborator

I have no idea if that's it (I have no idea about spark submit) but you have a typo: deploy-mode should be deploy_mode when i look at the code of spark_submit.py

2 replies

vmferreira81 Mar 3, 2022
Author

Thank you @potiuk , but this configuration works when I put it on the Connection configuration, under extras.

I have extended SparkSubmitHook. I set it to false where it checks if it is to tracker the driver status:

def _resolve_should_track_driver_status(self) -> bool: return False

This way, airflow doesn't try to poll the driver when it finished, giving to the task a SUCCESS flag. It was the way I managed to proceed.

potiuk Mar 7, 2022
Collaborator

Maybe you can provide a PR to that then, if you have the way to fix it ?

jornathanJ · 2024-03-25T06:48:03Z

jornathanJ
Mar 25, 2024

I hope this help someone like me :D

In case of me, I'm using

Airflow: 2.6.1
SparkSubmitOperator: 4.1.0
pyspark: 3.4.1
Spark 3.4.1

Set spark.master.rest.enabled=true to spark-default.conf.
After changing, we can see as below on spark master's web UI.

URL: spark://spark-master:7077
REST URL: spark://spark-master:6066 (cluster mode)

Go to Airflow > connection > edit: spark-default

connection Type: Spark <-- (updated: 2024-09-27) Do not type string, if you install package correctly you can find it at drop-down,
host: spark://spark-master
port: 6066
extra : {"spark-binary": "spark-submit", "deploy-mode": "cluster"} <-- (updated: 2024-09-27)

Now I can see log on dag's log as below.(AIRFLOW__LOGGING__LOGGING_LEVEL: 'DEBUG')

[2024-03-25, 15:03:36 KST] {spark_submit.py:559} DEBUG - polling status of spark driver with id driver-20240325060331-0000
[2024-03-25, 15:03:36 KST] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240325060331-0000']
[2024-03-25, 15:03:36 KST] {spark_submit.py:382} DEBUG - Poll driver status cmd: ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240325060331-0000']
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: Dload  Upload   Total   Spent    Left  Speed
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: 
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: 0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: 100   272  100   272    0     0  45333      0 --:--:-- --:--:-- --:--:-- 45333
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: {
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "action" : "SubmissionStatusResponse",
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "driverState" : "FAILED",
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "serverSparkVersion" : "3.4.1",
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "submissionId" : "driver-20240325060331-0000",
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "success" : true,
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "workerHostPort" : "172.28.0.2:17141",
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: "workerId" : "worker-20240325060303-172.28.0.2-17141"
[2024-03-25, 15:03:36 KST] {spark_submit.py:513} DEBUG - spark driver status log: }

And please double check on your dag file;

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

# we can set interval second via "status_poll_interval"
spark_submit = SparkSubmitOperator(
    ....
    conf=spark_conf,
    status_poll_interval=5,
)

4 replies

Jaideep-C Sep 5, 2024

But it wont be able to submit the jar to spark://spark-master:6066 right?
Can you share the spark-submit command logs?

jornathanJ Sep 27, 2024

It's too late, but I leave comment for someone who like encounter this issue.

Yes, spark submit doesn't provide feature like send jar to Spark cluster, we need prepare shared directory that access spark master and worker with same path or put it at class path.
Also you can use option like "--packages"

At previous, there was missing config, I updated "connection Type" and "extra".

I grep this log from Airflow UI and my Airflow, Spark is running on docker compose(spark docker image : bitnami/spark:3.5.1).
Please make sure, curl is in at airflow worker container(or node).
After submit job Operator watching status of Spark Job via curl. That's a reason why I need to install curl.

*** Found local files:
***   * /opt/bitnami/airflow/logs/dag_id=spark-submit-example/run_id=manual__2024-09-27T02:45:49.677077+00:00/task_id=spark_submit_job/attempt=1.log
[2024-09-27, 02:45:51 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: spark-submit-example.spark_submit_job manual__2024-09-27T02:45:49.677077+00:00 [queued]>
[2024-09-27, 02:45:51 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: spark-submit-example.spark_submit_job manual__2024-09-27T02:45:49.677077+00:00 [queued]>
[2024-09-27, 02:45:51 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2024-09-27, 02:45:51 UTC] {taskinstance.py:1327} INFO - Executing <Task(SparkSubmitOperator): spark_submit_job> on 2024-09-27 02:45:49.677077+00:00
[2024-09-27, 02:45:51 UTC] {standard_task_runner.py:57} INFO - Started process 1772 to run task
[2024-09-27, 02:45:51 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'spark-submit-example', 'spark_submit_job', 'manual__2024-09-27T02:45:49.677077+00:00', '--job-id', '4274', '--raw', '--subdir', 'DAGS_FOLDER/synopsys/data_lake_house/spark-submit-sample.py', '--cfg-path', '/tmp/tmpam_5heko']
[2024-09-27, 02:45:51 UTC] {standard_task_runner.py:85} INFO - Job 4274: Subtask spark_submit_job
[2024-09-27, 02:45:51 UTC] {warnings.py:109} WARNING - /opt/bitnami/airflow/venv/lib/python3.9/site-packages/airflow/settings.py:195: DeprecationWarning: The sql_alchemy_conn option in [core] has been moved to the sql_alchemy_conn option in [database] - the old setting has been used, but please update your config.
  SQL_ALCHEMY_CONN = conf.get("database", "SQL_ALCHEMY_CONN")
[2024-09-27, 02:45:51 UTC] {task_command.py:410} INFO - Running <TaskInstance: spark-submit-example.spark_submit_job manual__2024-09-27T02:45:49.677077+00:00 [running]> on host airflow-worker
[2024-09-27, 02:45:51 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='spark-submit-example' AIRFLOW_CTX_TASK_ID='spark_submit_job' AIRFLOW_CTX_EXECUTION_DATE='2024-09-27T02:45:49.677077+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-09-27T02:45:49.677077+00:00'
[2024-09-27, 02:45:51 UTC] {base.py:73} INFO - Using connection ID 'spark_default' for task execution.
[2024-09-27, 02:45:51 UTC] {spark_submit.py:339} INFO - Spark-Submit cmd: spark-submit --master spark://spark-master:6066 --conf spark.standalone.submit.waitAppCompletion=false --conf spark.driver.extraJavaOptions=-Divy.cache.dir=/opt/bitnami/spark/tmp -Divy.home=/opt/bitnami/spark/tmp --add-exports java.base/sun.nio.ch=ALL-UNNAMED --name airflow-spark-submit-1 --class org.apache.spark.examples.SparkPi --verbose --deploy-mode cluster /opt/bitnami/spark/examples/jars/spark-examples_2.12-3.5.1.jar
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Using properties file: /opt/spark/conf/spark-defaults.conf
--skipped
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Adding default property: spark.master.rest.enabled=true
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Adding default property: spark.driver.extraJavaOptions=--add-exports java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Adding default property: spark.standalone.submit.waitAppCompletion=true
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Parsed arguments:
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - master                  spark://spark-master:6066
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - remote                  null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - deployMode              cluster
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - executorMemory          null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - executorCores           null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - totalExecutorCores      null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - propertiesFile          /opt/spark/conf/spark-defaults.conf
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - driverMemory            null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - driverCores             null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - driverExtraClassPath    null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - driverExtraLibraryPath  null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - driverExtraJavaOptions  -Divy.cache.dir=/opt/bitnami/spark/tmp -Divy.home=/opt/bitnami/spark/tmp --add-exports java.base/sun.nio.ch=ALL-UNNAMED
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - supervise               false
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - queue                   null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - numExecutors            null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - files                   null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - pyFiles                 null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - archives                null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - mainClass               org.apache.spark.examples.SparkPi
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - primaryResource         file:/opt/bitnami/spark/examples/jars/spark-examples_2.12-3.5.1.jar
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - name                    airflow-spark-submit-1
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - childArgs               []
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - jars                    null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - packages                null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - packagesExclusions      null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - repositories            null
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - verbose                 true
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - 
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Spark properties used, including those specified through
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - --conf and those from the properties file /opt/spark/conf/spark-defaults.conf:
--skipped
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - 
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Running Spark using the REST application submission protocol.
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:53 DEBUG Shell:519 - Failed to detect a valid hadoop home directory
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset.
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - at org.apache.hadoop.util.Shell.checkHadoopHomeInner(Shell.java:467) ~[hadoop-client-api-3.3.4.jar:?]
--skipped
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:53 DEBUG Shell:820 - setsid exited with exit code 0
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Main class:
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - org.apache.spark.deploy.rest.RestSubmissionClientApp
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Arguments:
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - file:/opt/bitnami/spark/examples/jars/spark-examples_2.12-3.5.1.jar
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - org.apache.spark.examples.SparkPi
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Spark config:
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.app.name,airflow-spark-submit-1)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.app.submitTime,1727405153943)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.driver.extraJavaOptions,-Divy.cache.dir=/opt/bitnami/spark/tmp -Divy.home=/opt/bitnami/spark/tmp --add-exports java.base/sun.nio.ch=ALL-UNNAMED)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.driver.supervise,false)
--skipped
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.standalone.submit.waitAppCompletion,false)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.submit.deployMode,cluster)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - (spark.submit.pyFiles,)
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - Classpath elements:
[2024-09-27, 02:45:53 UTC] {spark_submit.py:490} INFO - 
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - Submitting a request to launch an application in spark://spark-master:6066.
[2024-09-27, 02:45:54 UTC] {spark_submit.py:488} INFO - identified spark driver id: driver-20240927024554-0003
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - Submission successfully created as driver-20240927024554-0003. Polling submission state...
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - Submitting a request for the status of submission driver-20240927024554-0003 in spark://spark-master:6066.
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - State of driver driver-20240927024554-0003 is now RUNNING.
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - Driver is running on worker worker-20240926042429-{workerIP}-39669 at {workerIP}:39669.
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  RestSubmissionClient:60 - Server responded with CreateSubmissionResponse:
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - {
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - "action" : "CreateSubmissionResponse",
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - "message" : "Driver successfully submitted as driver-20240927024554-0003",
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - "serverSparkVersion" : "3.5.1",
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - "submissionId" : "driver-20240927024554-0003",
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - "success" : true
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - }
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  ShutdownHookManager:60 - Shutdown hook called
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-c6444dea-056c-4d5a-bf77-59796d3688d0
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 DEBUG ShutdownHookManager:97 - Completed shutdown in 0.018 seconds; Timeouts: 0
[2024-09-27, 02:45:54 UTC] {spark_submit.py:490} INFO - 2024-09-27 02:45:54 DEBUG ShutdownHookManager:154 - ShutdownHookManager completed shutdown.
[2024-09-27, 02:45:55 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:45:56 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:45:57 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:45:58 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:00 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:01 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:02 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:03 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:04 UTC] {spark_submit.py:359} INFO - ['/usr/bin/curl', '--max-time', '30', 'http://spark-master:6066/v1/submissions/status/driver-20240927024554-0003']
[2024-09-27, 02:46:04 UTC] {taskinstance.py:1345} INFO - Marking task as SUCCESS. dag_id=spark-submit-example, task_id=spark_submit_job, execution_date=20240927T024549, start_date=20240927T024551, end_date=20240927T024604
[2024-09-27, 02:46:04 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2024-09-27, 02:46:04 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check

Here is connection config image

jornathanJ Sep 27, 2024

And this is simple DAG

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.utils.dates import days_ago

default_args = {
  "owner": "airflow",
  "start_date": datetime(2024, 9, 26)
}
dag_spark = DAG(
  dag_id = 'spark-submit-example',
  default_args=default_args, 
  schedule_interval='@once',
  dagrun_timeout=timedelta(minutes=60),
  description='use case of sparkoperator in airflow',  
)

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


operator = SparkSubmitOperator(
    task_id='spark_submit_job',
    conn_id='spark_default',
    java_class='org.apache.spark.examples.SparkPi',
    application='/opt/bitnami/spark/examples/jars/spark-examples_2.12-3.5.1.jar',
    name='airflow-spark-submit-1',
    verbose=True,
    conf=spark_conf,     
    dag=dag_spark
)

MeltonSmith Jan 22, 2025

I've faced the same issue recently
Airflow 2.10.4
Spark 3.5.4

I still don't get why 6066 is supposed to help :)

As the author says, the problem is in _resolve_should_track_driver_status method, which looks like this:

def _resolve_should_track_driver_status(self):
        """
        Determines whether or not this hook should poll the spark driver status through
        subsequent spark-submit status requests after the initial spark-submit request
        :return: if the driver status should be tracked
        """
        return ('spark://' in self._connection['master'] and
                self._connection['deploy_mode'] == 'cluster')

So I changed the type of spark connection in "connection" section of airflow from "Spark" to "Spark Connect"

Dag brings nothing extraordinary:

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

from airflow.operators.bash import BashOperator

from airflow.operators.python_operator import PythonOperator

from airflow.hooks.S3_hook import S3Hook

from datetime import datetime, timedelta

from textwrap import dedent

from airflow import DAG


s3_log_path = "s3a://spark-history-server/history"

spark_config = {

    "spark.sql.shuffle.partitions": 8,

    "spark.executor.memory":"4G",

    "spark.driver.memory":"4G",

    "spark.submit.deployMode": "cluster",

    "spark.standalone.submit.waitAppCompletion": "true",

    "spark.hadoop.fs.s3a.endpoint": "",

    "spark.hadoop.fs.s3a.access.key":"",

    "spark.hadoop.fs.s3a.secret.key":"",
}


with DAG(

    'foo',

    default_args={

        'depends_on_past': False,

        'retries': 1,

        'retry_delay': timedelta(minutes=5),

    },

    description='Some desc',

    schedule_interval=timedelta(days=1),

    start_date=datetime(2021, 1, 1),

    catchup=False,

    tags=['example'],

) as dag:

    t1 = SparkSubmitOperator(

        application="s3a://bucket/artifacts/driver/2.1.0/driver-2.1.0.jar",

        java_class="melt.smith.engine.driver.DriverApp",

        conf = spark_config,

        conn_id = "spark_default",

        task_id="some task name",

    )

hunnguye · 2026-08-18T11:36:49Z

hunnguye
Aug 18, 2026

This is still an issue in Airflow 3.2. also described in issue #20949, where I added some additional details.

If not planned yet, with some guidance, I’d be happy to try working on a fix for this.

0 replies

Uh oh!

Spark-submit task fails on deploy mode cluster #21799

Uh oh!

Uh oh!

vmferreira81 Feb 24, 2022

Replies: 3 comments · 6 replies

Uh oh!

Uh oh!

potiuk Feb 27, 2022 Collaborator

Uh oh!

Uh oh!

vmferreira81 Mar 3, 2022 Author

Uh oh!

potiuk Mar 7, 2022 Collaborator

Uh oh!

Uh oh!

jornathanJ Mar 25, 2024

Uh oh!

Jaideep-C Sep 5, 2024

Uh oh!

Uh oh!

jornathanJ Sep 27, 2024

Uh oh!

Uh oh!

jornathanJ Sep 27, 2024

Uh oh!

MeltonSmith Jan 22, 2025

Uh oh!

hunnguye Aug 18, 2026

vmferreira81
Feb 24, 2022

Replies: 3 comments 6 replies

potiuk
Feb 27, 2022
Collaborator

vmferreira81 Mar 3, 2022
Author

potiuk Mar 7, 2022
Collaborator

jornathanJ
Mar 25, 2024

hunnguye
Aug 18, 2026