<p>Apache Airflow version: v2.1.1</p>
<p>Kubernetes version (if you are using kubernetes) (use kubectl version):-
Client Version: version.Info{Major:&quot;1&quot;, Minor:&quot;21&quot;, GitVersion:&quot;v1.21.2&quot;, GitCommit:&quot;092fbfbf53427de67cac1e9fa54aaa09a28371d7&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2021-06-16T12:52:14Z&quot;, GoVersion:&quot;go1.16.5&quot;, Compiler:&quot;gc&quot;, Platform:&quot;darwin/amd64&quot;}
Server Version: version.Info{Major:&quot;1&quot;, Minor:&quot;19+&quot;, GitVersion:&quot;v1.19.8-eks-96780e&quot;, GitCommit:&quot;96780e1b30acbf0a52c38b6030d7853e575bcdf3&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2021-03-10T21:32:29Z&quot;, GoVersion:&quot;go1.15.8&quot;, Compiler:&quot;gc&quot;, Platform:&quot;linux/amd64&quot;}</p>
<p>Environment: Development</p>
<p>Cloud provider or hardware configuration: AWS EKS
OS (e.g. from /etc/os-release):
Kernel (e.g. uname -a):
Install tools:
Others:
What happened:
I am not able to create SparkApplications on the Kubernetes cluster using SparkKubernetesOperator from Airflow DAG. I have hosted Airflow and Spark-operator on EKS. I have created a connection on Airflow to connect to the Kubernetes cluster by using &quot;in cluster configuration&quot;.
I am just running the sample application just to check the execution of spark on Kubernetes through Airflow.</p>
<p>Application YAML file:-</p>
<pre><code>apiVersion: &quot;sparkoperator.k8s.io/v1beta2&quot;
kind: SparkApplication
metadata:
  name: spark-pi-airflow
  namespace: spark-apps
spec:
  type: Scala
  mode: cluster
  image: &quot;gcr.io/spark-operator/spark:v3.1.1&quot;
  imagePullPolicy: Always
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: &quot;local:///opt/spark/examples/jars/spark-examples_2.12-3.1.1.jar&quot;
  sparkVersion: &quot;3.1.1&quot;
  restartPolicy:
    type: Never
  volumes:
    - name: &quot;test-volume&quot;
      hostPath:
        path: &quot;/tmp&quot;
        type: Directory
  driver:
    cores: 1
    coreLimit: &quot;1200m&quot;
    memory: &quot;512m&quot;
    labels:
      version: 3.1.1
    serviceAccount: spark
    volumeMounts:
      - name: &quot;test-volume&quot;
        mountPath: &quot;/tmp&quot;
  executor:
    cores: 1
    instances: 1
    memory: &quot;512m&quot;
    labels:
      version: 3.1.1
    volumeMounts:
      - name: &quot;test-volume&quot;
        mountPath: &quot;/tmp&quot;
</code></pre>
<p>Airflow DAG:-</p>
<pre><code>
from datetime import timedelta

# [START import_module]
# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG

# Operators; we need this to operate!
from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator
from airflow.providers.cncf.kubernetes.sensors.spark_kubernetes import SparkKubernetesSensor
from airflow.utils.dates import days_ago

# [END import_module]

# [START default_args]
# These args will get passed on to each operator
# You can override them on a per-task basis during operator initialization
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'max_active_runs': 1,
}
# [END default_args]

# [START instantiate_dag]

dag = DAG(
    'spark_pi_airflow',
    default_args=default_args,
    description='submit spark-pi as sparkApplication on kubernetes',
    schedule_interval=timedelta(days=1),
    start_date=days_ago(1),
)

t1 = SparkKubernetesOperator(
    task_id='spark_pi_submit',
    namespace=&quot;spark-apps&quot;,
    application_file=&quot;example_spark_kubernetes_spark_pi.yaml&quot;,
    kubernetes_conn_id=&quot;kubernetes_default&quot;,
    do_xcom_push=True,
    dag=dag,
)

t2 = SparkKubernetesSensor(
    task_id='spark_pi_monitor',
    namespace=&quot;spark-apps&quot;,
    application_name=&quot;{{ task_instance.xcom_pull(task_ids='spark_pi_submit')['metadata']['name'] }}&quot;,
    kubernetes_conn_id=&quot;kubernetes_default&quot;,
    dag=dag,
)
t1 &gt;&gt; t2
</code></pre>
<p>Error Message:-</p>
<pre><code>
[2021-07-12 10:18:46,629] {spark_kubernetes.py:67} INFO - Creating sparkApplication
[2021-07-12 10:18:46,662] {taskinstance.py:1501} ERROR - Task failed with exception
Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/hooks/kubernetes.py&quot;, line 174, in create_custom_object
    response = api.create_namespaced_custom_object(
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/api/custom_objects_api.py&quot;, line 183, in create_namespaced_custom_object
    (data) = self.create_namespaced_custom_object_with_http_info(group, version, namespace, plural, body, **kwargs)  # noqa: E501
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/api/custom_objects_api.py&quot;, line 275, in create_namespaced_custom_object_with_http_info
    return self.api_client.call_api(
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/api_client.py&quot;, line 340, in call_api
    return self.__call_api(resource_path, method,
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/api_client.py&quot;, line 172, in __call_api
    response_data = self.request(
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/api_client.py&quot;, line 382, in request
    return self.rest_client.POST(url,
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/rest.py&quot;, line 272, in POST
    return self.request(&quot;POST&quot;, url,
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/kubernetes/client/rest.py&quot;, line 231, in request
    raise ApiException(http_resp=r)
kubernetes.client.rest.ApiException: (403)
Reason: Forbidden
HTTP response headers: HTTPHeaderDict({'Audit-Id': '45712aa7-85e3-4beb-85f7-b94a77cda196', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'Date': 'Mon, 12 Jul 2021 10:18:46 GMT', 'Content-Length': '406'})
HTTP response body: {&quot;kind&quot;:&quot;Status&quot;,&quot;apiVersion&quot;:&quot;v1&quot;,&quot;metadata&quot;:{},&quot;status&quot;:&quot;Failure&quot;,&quot;message&quot;:&quot;sparkapplications.sparkoperator.k8s.io is forbidden: User \&quot;system:serviceaccount:airflow:airflow-cluster\&quot; cannot create resource \&quot;sparkapplications\&quot; in API group \&quot;sparkoperator.k8s.io\&quot; in the namespace \&quot;spark-apps\&quot;&quot;,&quot;reason&quot;:&quot;Forbidden&quot;,&quot;details&quot;:{&quot;group&quot;:&quot;sparkoperator.k8s.io&quot;,&quot;kind&quot;:&quot;sparkapplications&quot;},&quot;code&quot;:403}



During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py&quot;, line 1157, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py&quot;, line 1331, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py&quot;, line 1361, in _execute_task
    result = task_copy.execute(context=context)
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py&quot;, line 69, in execute
    response = hook.create_custom_object(
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/hooks/kubernetes.py&quot;, line 180, in create_custom_object
    raise AirflowException(f&quot;Exception when calling -&gt; create_custom_object: {e}\n&quot;)
airflow.exceptions.AirflowException: Exception when calling -&gt; create_custom_object: (403)
Reason: Forbidden
HTTP response headers: HTTPHeaderDict({'Audit-Id': '45712aa7-85e3-4beb-85f7-b94a77cda196', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'Date': 'Mon, 12 Jul 2021 10:18:46 GMT', 'Content-Length': '406'})
HTTP response body: {&quot;kind&quot;:&quot;Status&quot;,&quot;apiVersion&quot;:&quot;v1&quot;,&quot;metadata&quot;:{},&quot;status&quot;:&quot;Failure&quot;,&quot;message&quot;:&quot;sparkapplications.sparkoperator.k8s.io is forbidden: User \&quot;system:serviceaccount:***:***-cluster\&quot; cannot create resource \&quot;sparkapplications\&quot; in API group \&quot;sparkoperator.k8s.io\&quot; in the namespace \&quot;spark-apps\&quot;&quot;,&quot;reason&quot;:&quot;Forbidden&quot;,&quot;details&quot;:{&quot;group&quot;:&quot;sparkoperator.k8s.io&quot;,&quot;kind&quot;:&quot;sparkapplications&quot;},&quot;code&quot;:403}
</code></pre>
<p>What you expected to happen:
Kubernetes Airflow should schedule and run spark job using SparkKubernetesOperator.</p>
<p>How to reproduce it:
Deploy Spark operator using helm on Kubernetes cluster.
Deploy Airflow using helm on Kubernetes cluster.
Deploy the above-mentioned application and Airflow DAG.</p>
<p>Anything else we need to know:-</p>
<p>I have already created service account:-</p>
<pre><code>$ kubectl create serviceaccount spark
</code></pre>
<p>Given the service account the edit role on the cluster:-</p>
<pre><code>$ kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=airflow:airflow-cluster --namespace=spark-apps
</code></pre>

<p>Here are kube cluster role resources. Create with <code>kubectl -n &lt;namespace&gt; apply -f &lt;filename.yaml&gt;</code></p>
<pre class="lang-yaml prettyprint-override"><code># Role for spark-on-k8s-operator to create resources on cluster
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: spark-cluster-cr
  labels:
    rbac.authorization.kubeflow.org/aggregate-to-kubeflow-edit: &quot;true&quot;
rules:
  - apiGroups:
      - sparkoperator.k8s.io
    resources:
      - sparkapplications
    verbs:
      - '*'
---
# Allow airflow-worker service account access for spark-on-k8s
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: airflow-spark-crb
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: spark-cluster-cr
subjects:
  - kind: ServiceAccount
    name: airflow-cluster
    namespace: airflow
</code></pre>
<p>Notes:</p>
<ul>
<li>The above is assuming the error message
<pre><code>sparkapplications.sparkoperator.k8s.io is forbidden: User &quot;system:serviceaccount:airflow:airflow-cluster\&quot; cannot create resource \&quot;sparkapplications\&quot; in API group \&quot;sparkoperator.k8s.io\&quot; in the namespace spark-apps
</code></pre>
<ul>
<li>Airflow namespace: <code>airflow</code></li>
<li>Airflow serviceaccount: <code>airflow-cluster</code></li>
<li>Spark jobs namespace: <code>spark-apps</code></li>
</ul>
</li>
<li>You should also have <a href="https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/master/docs/quick-start-guide.md" rel="noreferrer">spark-on-k8s-operator</a> installed
<ul>
<li>With helm <code> --set webhook.enable=true</code> if you want to use <code>env</code> in your <code>spec.driver</code></li>
</ul>
</li>
</ul>
