<p>I'm running Airflow on Kubernetes using this Helm chart: <a href="https://github.com/apache/airflow/tree/1.5.0" rel="nofollow noreferrer">https://github.com/apache/airflow/tree/1.5.0</a></p>
<p>I've written a very simple DAG just to test some things. It looks like this:</p>
<pre><code>default_args={
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'my-dag',
    default_args=default_args,
    description='simple dag',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2022, 4, 21),
    catchup=False,
    tags=['example']
) as dag:

    t1 = SparkKubernetesOperator(
        task_id='spark-pi',
        trigger_rule=&quot;all_success&quot;,
        depends_on_past=False,
        retries=3,
        application_file=&quot;spark-pi.yaml&quot;,
        namespace=&quot;my-ns&quot;,
        kubernetes_conn_id=&quot;myk8s&quot;,
        api_group=&quot;sparkoperator.k8s.io&quot;,
        api_version=&quot;v1beta2&quot;,
        do_xcom_push=True,
        dag=dag
    )

    t2 = SparkKubernetesOperator(
        task_id='other-spark-job',
        trigger_rule=&quot;all_success&quot;,
        depends_on_past=False,
        retries=3,
        application_file=other-spark-job-definition,
        namespace=&quot;my-ns&quot;,
        kubernetes_conn_id=&quot;myk8s&quot;,
        api_group=&quot;sparkoperator.k8s.io&quot;,
        api_version=&quot;v1beta2&quot;,
        dag=dag
    )

    t1 &gt;&gt; t2
</code></pre>
<p>When I run the DAG from the Airflow UI, the first task Spark job (<code>t1</code>, spark-pi) gets created and is immediately marked as successful, and then Airflow launches the second (t2) task right after that. This can be seen in the web UI:</p>
<p><a href="https://i.stack.imgur.com/CXBP3.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/CXBP3.png" alt="enter image description here" /></a></p>
<p>What you're seeing is the status of the two tasks in 5 separate DAG runs, as well as their total status (the circles). The middle row of the image shows the status of <code>t1</code>, which is <code>&quot;success&quot;</code>.</p>
<p><strong>However</strong>, the actual spark-pi pod of <code>t1</code> launched by the Spark operator fails on every run, and its status can be seen by querying the Sparkapplication resource on Kubernetes:</p>
<pre><code>$ kubectl get sparkapplications/spark-pi-2022-04-28-2 -n my-ns -o json
{
    &quot;apiVersion&quot;: &quot;sparkoperator.k8s.io/v1beta2&quot;,
    &quot;kind&quot;: &quot;SparkApplication&quot;,
    &quot;metadata&quot;: {
        &quot;creationTimestamp&quot;: &quot;2022-04-29T13:28:02Z&quot;,
        &quot;generation&quot;: 1,
        &quot;name&quot;: &quot;spark-pi-2022-04-28-2&quot;,
        &quot;namespace&quot;: &quot;my-ns&quot;,
        &quot;resourceVersion&quot;: &quot;111463226&quot;,
        &quot;uid&quot;: &quot;23f1c8fb-7843-4628-b22f-7808b562f9d8&quot;
    },
    &quot;spec&quot;: {
        &quot;driver&quot;: {
            &quot;coreLimit&quot;: &quot;1500m&quot;,
            &quot;cores&quot;: 1,
            &quot;labels&quot;: {
                &quot;version&quot;: &quot;2.4.4&quot;
            },
            &quot;memory&quot;: &quot;512m&quot;,
            &quot;volumeMounts&quot;: [
                {
                    &quot;mountPath&quot;: &quot;/tmp&quot;,
                    &quot;name&quot;: &quot;test-volume&quot;
                }
            ]
        },
        &quot;executor&quot;: {
            &quot;coreLimit&quot;: &quot;1500m&quot;,
            &quot;cores&quot;: 1,
            &quot;instances&quot;: 1,
            &quot;labels&quot;: {
                &quot;version&quot;: &quot;2.4.4&quot;
            },
            &quot;memory&quot;: &quot;512m&quot;,
            &quot;volumeMounts&quot;: [
                {
                    &quot;mountPath&quot;: &quot;/tmp&quot;,
                    &quot;name&quot;: &quot;test-volume&quot;
                }
            ]
        },
        &quot;image&quot;: &quot;my.google.artifactory.com/spark-operator/spark:v2.4.4&quot;,
        &quot;imagePullPolicy&quot;: &quot;Always&quot;,
        &quot;mainApplicationFile&quot;: &quot;local:///opt/spark/examples/jars/spark-examples_2.11-2.4.4.jar&quot;,
        &quot;mainClass&quot;: &quot;org.apache.spark.examples.SparkPi&quot;,
        &quot;mode&quot;: &quot;cluster&quot;,
        &quot;restartPolicy&quot;: {
            &quot;type&quot;: &quot;Never&quot;
        },
        &quot;sparkVersion&quot;: &quot;2.4.4&quot;,
        &quot;type&quot;: &quot;Scala&quot;,
        &quot;volumes&quot;: [
            {
                &quot;hostPath&quot;: {
                    &quot;path&quot;: &quot;/tmp&quot;,
                    &quot;type&quot;: &quot;Directory&quot;
                },
                &quot;name&quot;: &quot;test-volume&quot;
            }
        ]
    },
    &quot;status&quot;: {
        &quot;applicationState&quot;: {
            &quot;errorMessage&quot;: &quot;driver container failed with ExitCode: 1, Reason: Error&quot;,
            &quot;state&quot;: &quot;FAILED&quot;
        },
        &quot;driverInfo&quot;: {
            &quot;podName&quot;: &quot;spark-pi-2022-04-28-2-driver&quot;,
            &quot;webUIAddress&quot;: &quot;172.20.23.178:4040&quot;,
            &quot;webUIPort&quot;: 4040,
            &quot;webUIServiceName&quot;: &quot;spark-pi-2022-04-28-2-ui-svc&quot;
        },
        &quot;executionAttempts&quot;: 1,
        &quot;lastSubmissionAttemptTime&quot;: &quot;2022-04-29T13:28:15Z&quot;,
        &quot;sparkApplicationId&quot;: &quot;spark-3335e141a51148d7af485457212eb389&quot;,
        &quot;submissionAttempts&quot;: 1,
        &quot;submissionID&quot;: &quot;021e78fc-4754-4ac8-a87d-52c682ddc483&quot;,
        &quot;terminationTime&quot;: &quot;2022-04-29T13:28:25Z&quot;
    }
}
</code></pre>
<p>As you can see in the <code>status</code> section, we have <code>&quot;state&quot;: &quot;FAILED&quot;</code>. Still, Airflow marks it as successful and thus runs t2 right after it, which is not what we want when defining <code>t2</code> as dependent on (downstream of) <code>t1</code>.</p>
<p>Why does Airflow see <code>t1</code> as successful even though the Spark job itself fails?</p>

Sahand
<p>That's the implementation. If you see the code for the operator it is basically a submit and forget job. To monitor the status you use SparkkubernetesSensor</p>
<pre class="lang-py prettyprint-override"><code>       t2 = SparkKubernetesSensor(
            task_id=&quot;spark_monitor&quot;,
            application_name=&quot;{{ task_instance.xcom_pull(task_ids='spark-job-full-refresh.spark_full_refresh') ['metadata']['name'] }}&quot;,
            attach_log=True,
        )
</code></pre>
<p>I have tried to create a custom operator that combines both but it does not work very well via inheritance because they are slightly different execution patterns, so it needs to be created from scratch. But for all purposes and intents, the Sensor works perfectly, just adds unneeded lines to code.</p>
