working dag

Author: adwk67

demos/demos-v2.yaml

@@ -165,6 +165,7 @@ demos:
       - iceberg
       - minio
       - s3
+      - airflow
     manifests: []
     supportedNamespaces: []
   trino-taxi-data:

stacks/stacks-v2.yaml

@@ -394,11 +394,12 @@ stacks:
     manifests:
       - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/_templates/minio-distributed-small-tls/rendered-chart.yaml
       - helmChart: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/_templates/postgresql-hive-iceberg.yaml
+      - helmChart: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/_templates/postgresql-airflow.yaml
       - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/trino-iceberg/s3-connection.yaml
       - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/trino-iceberg/hive-metastores.yaml
       - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/trino-iceberg/trino.yaml
-      - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/stacks/_templates/postgresql-airflow.yaml
       - plainYaml: stacks/trino-iceberg/airflow.yaml
+      - helmChart: stacks/trino-iceberg/redis-airflow.yaml
     supportedNamespaces: ["default"]
     parameters:
       - name: trinoAdminPassword

stacks/trino-iceberg/airflow.yaml

@@ -5,42 +5,87 @@ metadata:
   name: airflow
 spec:
   image:
-    productVersion: 3.0.6
+    productVersion: 2.10.5
     pullPolicy: IfNotPresent
   clusterConfig:
     loadExamples: false
     credentialsSecret: airflow-credentials
+    volumes:
+      - name: airflow-dags
+        configMap:
+          name: airflow-dags
+    volumeMounts:
+      - name: airflow-dags
+        mountPath: /stackable/airflow/dags/dbt_trino.py
+        subPath: dbt_trino.py
   webservers:
     roleConfig:
       listenerClass: external-stable
+    envOverrides: &envOverrides
+      TZ: "Europe/Berlin"
+      #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"
+      #AIRFLOW_CONN_KUBERNETES_DEFAULT: "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"
     config:
       resources:
         cpu:
           min: "2"
           max: "3"
         memory:
           limit: 3Gi
-    configOverrides:
-      webserver_config.py:
-        # Allow "POST /login/" without CSRF token
-        WTF_CSRF_ENABLED: "False"
     roleGroups:
       default:
         replicas: 1
-  kubernetesExecutors:
-    config: {}
-  schedulers:
-    roleGroups:
-      default:
-        replicas: 1
-  dagProcessors:
+  celeryExecutors:
+    envOverrides: *envOverrides
     roleGroups:
       default:
         replicas: 1
-  triggerers:
+  schedulers:
+    envOverrides: *envOverrides
     roleGroups:
       default:
         replicas: 1
+  # dagProcessors:
+  #   envOverrides: *envOverrides
+  #   roleGroups:
+  #     default:
+  #       replicas: 1
+  # triggerers:
+  #   envOverrides: *envOverrides
+  #   roleGroups:
+  #     default:
+  #       replicas: 1
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: airflow-dags
+data:
+  dbt_trino.py: |
+    from airflow import DAG
+    from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
+    import pendulum
+    from datetime import datetime
+
+    with DAG(
+        dag_id="run_dbt_check_via_pod_operator",
+        schedule=None,
+        start_date=pendulum.datetime(2024, 9, 1),
+        tags=["Demo", "DBT"],
+        catchup=False
+    ) as dag:
+      run_dbt_check = KubernetesPodOperator(
+          namespace="default",
+          image="my-dbt-trino:0.0.1",
+          image_pull_policy="IfNotPresent",
+          cmds=["/bin/bash", "-x", "-euo", "pipefail", "-c"],
+          arguments=["dbt debug"],
+          name="run-dbt-debug",
+          task_id="task",
+          #on_finish_action="keep_pod",
+          startup_timeout_seconds=600
+      )
+      run_dbt_check
 ---
 apiVersion: v1
 kind: Secret
@@ -55,3 +100,5 @@ stringData:
   adminUser.password: "adminadmin"
   connections.secretKey: "airflowSecretKey"
   connections.sqlalchemyDatabaseUri: postgresql+psycopg2://airflow:airflow@postgresql-airflow/airflow
+  connections.celeryResultBackend: db+postgresql://airflow:airflow@postgresql-airflow/airflow
+  connections.celeryBrokerUrl: redis://:redis@redis-airflow-master:6379/0

stacks/trino-iceberg/redis-airflow.yaml

@@ -0,0 +1,25 @@
+---
+releaseName: redis-airflow
+name: redis
+repo:
+  name: bitnami
+  url: https://charts.bitnami.com/bitnami/
+version: 17.11.3
+options:
+  commonLabels:
+    stackable.tech/vendor: Stackable
+  auth:
+    password: redis
+  global:
+    security:
+      allowInsecureImages: true
+  image:
+    repository: bitnamilegacy/redis
+  volumePermissions:
+    image:
+      repository: bitnamilegacy/os-shell
+  metrics:
+    image:
+      repository: bitnamilegacy/redis-exporter
+  replica:
+    replicaCount: 1