Usage

webservers: envOverrides: AIRFLOW__WEBSERVER__AUTO_REFRESH_INTERVAL: "8" roleGroups: default: replicas: 1

webservers: roleGroups: default: envOverrides: AIRFLOW__WEBSERVER__AUTO_REFRESH_INTERVAL: "8" replicas: 1

--- apiVersion: example.stackable.tech/v1alpha1 kind: ExampleCluster metadata: name: example spec: workers: # role-level config: resources: cpu: min: 300m max: 600m memory: limit: 3Gi roleGroups: # role-group-level resources-from-role: # role-group 1 replicas: 1 resources-from-role-group: # role-group 2 replicas: 1 config: resources: cpu: min: 400m max: 800m memory: limit: 4Gi

workers: roleGroups: default: config: resources: cpu: min: '200m' max: "4" memory: limit: '2Gi'

apiVersion: airflow.stackable.tech/v1alpha1 kind: AirflowCluster metadata: name: airflow-with-ldap spec: image: productVersion: 2.4.1 stackableVersion: 0.7.0 [...] authenticationConfig: authenticationClass: ldap (1) userRegistrationRole: Admin (2)

apiVersion: airflow.stackable.tech/v1alpha1 kind: AirflowCluster metadata: name: airflow-with-ldap spec: [...] authenticationConfig: authenticationClass: ldap userRegistrationRole: Admin (1)

--- apiVersion: v1 kind: ConfigMap metadata: name: cm-dag (1) data: test_airflow_dag.py: | (2) from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.dummy import DummyOperator with DAG( dag_id='test_airflow_dag', schedule_interval='0 0 * * *', start_date=datetime(2021, 1, 1), catchup=False, dagrun_timeout=timedelta(minutes=60), tags=['example', 'example2'], params={"example_key": "example_value"}, ) as dag: run_this_last = DummyOperator( task_id='run_this_last', ) # [START howto_operator_bash] run_this = BashOperator( task_id='run_after_loop', bash_command='echo 1', ) # [END howto_operator_bash] run_this >> run_this_last for i in range(3): task = BashOperator( task_id='runme_' + str(i), bash_command='echo "{{ task_instance_key_str }}" && sleep 1', ) task >> run_this # [START howto_operator_bash_template] also_run_this = BashOperator( task_id='also_run_this', bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"', ) # [END howto_operator_bash_template] also_run_this >> run_this_last # [START howto_operator_bash_skip] this_will_skip = BashOperator( task_id='this_will_skip', bash_command='echo "hello world"; exit 99;', dag=dag, ) # [END howto_operator_bash_skip] this_will_skip >> run_this_last if __name__ == "__main__": dag.cli()

--- apiVersion: airflow.stackable.tech/v1alpha1 kind: AirflowCluster metadata: name: airflow spec: image: productVersion: 2.4.1 stackableVersion: 0.7.0 executor: CeleryExecutor loadExamples: false exposeConfig: false credentialsSecret: simple-airflow-credentials volumes: - name: cm-dag (3) configMap: name: cm-dag (4) volumeMounts: - name: cm-dag (5) mountPath: /dags/test_airflow_dag.py (6) subPath: test_airflow_dag.py (7) webservers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8) replicas: 1 workers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8) replicas: 2 schedulers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8) replicas: 1

--- apiVersion: v1 kind: PersistentVolumeClaim metadata: name: pvc-airflow (1) spec: accessModes: - ReadWriteOnce resources: requests: storage: 1Gi --- apiVersion: batch/v1 kind: Job (2) metadata: name: airflow-dags spec: template: spec: restartPolicy: Never volumes: - name: external-dags (3) persistentVolumeClaim: claimName: pvc-airflow (4) initContainers: - name: dest-dir image: docker.stackable.tech/stackable/tools:0.2.0-stackable0.4.0 env: - name: DEST_DIR value: "/stackable/externals" command: [ "bash", "-x", "-c", "mkdir -p $DEST_DIR && chown stackable:stackable ${DEST_DIR} && chmod -R a=,u=rwX,g=rwX ${DEST_DIR}", ] securityContext: runAsUser: 0 volumeMounts: - name: external-dags (5) mountPath: /stackable/externals (6) containers: - name: airflow-dags image: docker.stackable.tech/stackable/tools:0.2.0-stackable0.4.0 env: - name: DEST_DIR value: "/stackable/externals" command: (7) [ "bash", "-x", "-c", "curl -L https://raw.githubusercontent.com/apache/airflow/2.4.1/airflow/example_dags/example_bash_operator.py \ -o ${DEST_DIR}/example_bash_operator.py && \ curl -L https://raw.githubusercontent.com/apache/airflow/2.4.1/airflow/example_dags/example_complex.py \ -o ${DEST_DIR}/example_complex.py && \ curl -L https://raw.githubusercontent.com/apache/airflow/2.4.1/airflow/example_dags/example_branch_datetime_operator.py \ -o ${DEST_DIR}/example_branch_datetime_operator.py", ] volumeMounts: - name: external-dags (5) mountPath: /stackable/externals (6)

--- apiVersion: airflow.stackable.tech/v1alpha1 kind: AirflowCluster metadata: name: airflow spec: image: productVersion: 2.4.1 stackableVersion: 0.7.0 executor: CeleryExecutor loadExamples: false exposeConfig: false credentialsSecret: simple-airflow-credentials volumes: - name: external-dags (8) persistentVolumeClaim: claimName: pvc-airflow (9) volumeMounts: - name: external-dags (10) mountPath: /stackable/external-dags (11) webservers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12) replicas: 1 workers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12) replicas: 2 schedulers: roleGroups: default: envOverrides: AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12) replicas: 1

AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D"

--- apiVersion: airflow.stackable.tech/v1alpha1 kind: AirflowCluster metadata: name: airflow spec: image: productVersion: 2.4.1 stackableVersion: 0.7.0 executor: CeleryExecutor loadExamples: false exposeConfig: false credentialsSecret: simple-airflow-credentials webservers: roleGroups: default: envOverrides: AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D" replicas: 1 workers: roleGroups: default: envOverrides: AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D" replicas: 1 schedulers: roleGroups: default: envOverrides: AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D" replicas: 1

--- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: airflow-spark-clusterrole rules: - apiGroups: - spark.stackable.tech resources: - sparkapplications verbs: - create - get

--- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: airflow-spark-clusterrole-binding roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole name: airflow-spark-clusterrole subjects: - apiGroup: rbac.authorization.k8s.io kind: Group name: system:serviceaccounts

--- apiVersion: spark.stackable.tech/v1alpha1 kind: SparkApplication metadata: name: pyspark-pi namespace: default spec: version: "1.0" sparkImage: docker.stackable.tech/stackable/pyspark-k8s:3.3.0-stackable0.3.0 mode: cluster mainApplicationFile: local:///stackable/spark/examples/src/main/python/pi.py driver: cores: 1 coreLimit: "1200m" memory: "512m" executor: cores: 1 instances: 3 memory: "512m"

# # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """Example DAG demonstrating how to apply a Kubernetes Resource from Airflow running in-cluster""" from datetime import datetime, timedelta from airflow import DAG from typing import TYPE_CHECKING, Optional, Sequence, Dict from kubernetes import client from airflow.exceptions import AirflowException from airflow.sensors.base import BaseSensorOperator from airflow.models import BaseOperator from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook if TYPE_CHECKING: from airflow.utils.context import Context class SparkKubernetesOperator(BaseOperator): (1) """ Creates a SparkApplication resource in kubernetes: :param application_file: Defines a 'SparkApplication' custom resource as either a path to a '.yaml' file, '.json' file, YAML string or JSON string. :param namespace: kubernetes namespace for the SparkApplication :param kubernetes_conn_id: The :ref:`kubernetes connection id <howto/connection:kubernetes>` for the Kubernetes cluster. :param api_group: SparkApplication API group :param api_version: SparkApplication API version """ template_fields: Sequence[str] = ('application_file', 'namespace') template_ext: Sequence[str] = ('.yaml', '.yml', '.json') ui_color = '#f4a460' def __init__( self, *, application_file: str, namespace: Optional[str] = None, kubernetes_conn_id: str = 'kubernetes_in_cluster', (2) api_group: str = 'spark.stackable.tech', api_version: str = 'v1alpha1', **kwargs, ) -> None: super().__init__(**kwargs) self.application_file = application_file self.namespace = namespace self.kubernetes_conn_id = kubernetes_conn_id self.api_group = api_group self.api_version = api_version self.plural = "sparkapplications" def execute(self, context: 'Context'): hook = KubernetesHook(conn_id=self.kubernetes_conn_id) self.log.info("Creating SparkApplication...") response = hook.create_custom_object( group=self.api_group, version=self.api_version, plural=self.plural, body=self.application_file, namespace=self.namespace, ) return response class SparkKubernetesSensor(BaseSensorOperator): (3) """ Monitors a SparkApplication resource in kubernetes: :param application_name: SparkApplication resource name :param namespace: the kubernetes namespace where the SparkApplication reside in :param kubernetes_conn_id: The :ref:`kubernetes connection<howto/connection:kubernetes>` to Kubernetes cluster. :param attach_log: determines whether logs for driver pod should be appended to the sensor log :param api_group: SparkApplication API group :param api_version: SparkApplication API version """ template_fields = ("application_name", "namespace") FAILURE_STATES = ("Failed", "Unknown") SUCCESS_STATES = ("Succeeded") def __init__( self, *, application_name: str, attach_log: bool = False, namespace: Optional[str] = None, kubernetes_conn_id: str = 'kubernetes_in_cluster', (2) api_group: str = 'spark.stackable.tech', api_version: str = 'v1alpha1', poke_interval: float = 60, **kwargs, ) -> None: super().__init__(**kwargs) self.application_name = application_name self.attach_log = attach_log self.namespace = namespace self.kubernetes_conn_id = kubernetes_conn_id self.hook = KubernetesHook(conn_id=self.kubernetes_conn_id) self.api_group = api_group self.api_version = api_version self.poke_interval = poke_interval def _log_driver(self, application_state: str, response: dict) -> None: if not self.attach_log: return status_info = response["status"] if "driverInfo" not in status_info: return driver_info = status_info["driverInfo"] if "podName" not in driver_info: return driver_pod_name = driver_info["podName"] namespace = response["metadata"]["namespace"] log_method = self.log.error if application_state in self.FAILURE_STATES else self.log.info try: log = "" for line in self.hook.get_pod_logs(driver_pod_name, namespace=namespace): log += line.decode() log_method(log) except client.rest.ApiException as e: self.log.warning( "Could not read logs for pod %s. It may have been disposed.\n" "Make sure timeToLiveSeconds is set on your SparkApplication spec.\n" "underlying exception: %s", driver_pod_name, e, ) def poke(self, context: Dict) -> bool: self.log.info("Poking: %s", self.application_name) response = self.hook.get_custom_object( group=self.api_group, version=self.api_version, plural="sparkapplications", name=self.application_name, namespace=self.namespace, ) try: application_state = response["status"]["phase"] except KeyError: self.log.debug(f"SparkApplication status could not be established: {response}") return False if self.attach_log and application_state in self.FAILURE_STATES + self.SUCCESS_STATES: self._log_driver(application_state, response) if application_state in self.FAILURE_STATES: raise AirflowException(f"SparkApplication failed with state: {application_state}") elif application_state in self.SUCCESS_STATES: self.log.info("SparkApplication ended successfully") return True else: self.log.info("SparkApplication is still in state: %s", application_state) return False with DAG( (4) dag_id='sparkapp_dag', schedule_interval='0 0 * * *', start_date=datetime(2021, 1, 1), catchup=False, dagrun_timeout=timedelta(minutes=60), tags=['example'], params={"example_key": "example_value"}, ) as dag: t1 = SparkKubernetesOperator( (5) task_id='spark_pi_submit', namespace="default", application_file="pyspark-pi.yaml", do_xcom_push=True, dag=dag, ) t2 = SparkKubernetesSensor( (6) task_id='spark_pi_monitor', namespace="default", application_name="{{ task_instance.xcom_pull(task_ids='spark_pi_submit')['metadata']['name'] }}", poke_interval=5, dag=dag, ) t1 >> t2 (7)