diff --git a/dags_airflow_ale_trainmod.py b/dags_airflow_ale_trainmod.py index 4f4460d..e17c824 100644 --- a/dags_airflow_ale_trainmod.py +++ b/dags_airflow_ale_trainmod.py @@ -5,6 +5,8 @@ from airflow.providers.http.sensors.http import HttpSensor from airflow.providers.http.operators.http import SimpleHttpOperator from airflow.operators.python import BranchPythonOperator +from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator +from kubernetes.client import models as k8s import json from datetime import timedelta @@ -122,11 +124,50 @@ def check_response(response): join_branch = DummyOperator(task_id="join_brc",trigger_rule="none_failed", dag=dag) -list_proc = DummyOperator(task_id="listen_preproc", dag=dag) + + +preproc =KubernetesPodOperator( + # unique id of the task within the DAG + task_id="preQ", + # the Docker image to launch + image="romanoa77/preq:0.2", + # launch the Pod on the same cluster as Airflow is running on + in_cluster=True, + # launch the Pod in the same namespace as Airflow is running in + namespace="glitchflow", + # Pod configuration + # name the Pod + name="airflow_preprocessor", + + # attach labels to the Pod, can be used for grouping + labels={"app": "preq", "backend": "airflow"}, + # reattach to worker instead of creating a new Pod on worker failure + reattach_on_restart=True, + # delete Pod after the task is finished + is_delete_operator_pod=True, + # get log stdout of the container as task logs + get_logs=True, + # log events in case of Pod failure + log_events_on_failure=True, + # enable xcom + do_xcom_push=True, + volumes = k8s.V1Volume( + name="gwdatal", + persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name="gwdatal"), + ), + volume_mounts=[ + k8s.V1VolumeMount(mount_path="/app/data", name="gwdatal", sub_path=None, read_only=False) + ], + dag=dag, + + #env_vars={"NAME_TO_GREET": f"{name}"}, + ) + + IniTrain>>sign_train>>chech_train_resp>>[next_sens,next_metrics] next_sens>>freeze [next_metrics,freeze]>>join_branch -join_branch>>list_proc \ No newline at end of file +join_branch>>preproc \ No newline at end of file