黑皮考町

如何重跑 Cloud Composer 上的任務 Retry Cloud Composer Task  想要重跑 Cloud Composer 上的任務可以透過 Airflow Cli 來清除任務狀態, 迫使 Airflow 重新排程這個任務 清除指令, 如下 airflow tasks clear -s ${EXECUTION_DATE} -t ${TASK_ID} -d -y ${DAG_ID}   註：EXECUTION_DATE 的格式是以 ISO-8601 來表示, 所以會長這樣 "2022-07-26T00:00:00+00:00" 由於 Cloud Composer 會將 Airflow 部署在 GKE 上, 若要執行上面的指令, 必須 1. 取得 GKE 得 crendential 2. 找到 Airflow 管理工具的 Pod 執行 Airflow 指令 另一個方法是 使用 gcloud composer 來將指令傳給 Airflow  EXECUTION_DATE ="2022-07-26T00:00:00+00:00" DAG_ID="sysops-daily-backup" TASK_ID ="^result_chk$" gcloud composer environments run \ --location="us-central1" \ --project="stage" sysops tasks clear -- {DAG_ID} -s {EXECUTION_DATE}  -t {TASK_ID} -y -d  -d 是連下游任務的狀態都清除 方法二：在 AirFlow DAG 中使用 retry 機制來清除上游任務的狀態 def clear_upstream_task(context):     execution_date = context.get("execution_date")     clear_tasks = BashOperator(         task_id='clear_tasks',         bash_command=f'...

如何讓在 Airflow 上跑的任務使用 GPU 資源 Step 1. 在既有的 Airflow 下新增有 GPU 的 Pool 使用參數 --accelerator 指定 GPU 的規格 gcloud beta container   node-pools create "gpu-pool" \ --cluster "mlworkflow-24aeef46-gke" \ --zone "us-east1-b" \ --machine-type "n1-standard-4" \ --accelerator "type=nvidia-tesla-p100,count=1" \ --image-type "COS_CONTAINERD" \ --num-nodes "3"  Step 2. 測試 GPU 的功能是否正常 部署一個 Pod 來跑 nvidia-smi 指令 注意: GCP 在建立 GPU 的 Pool 時會自動加上 Taint 資訊如下  spec:   taints:   - effect: NoSchedule     key: nvidia.com/gpu     value: present Pod  的 deployment 要加上 Tolerations cat <<EOF | kubectl apply -f - apiVersion: v1 kind: Pod metadata:   labels:     run: mig-none-example   name: mig-none-example spec:   tolerations:   - key: "nvidia.com/gpu"     operator: "Exists"     effect: "NoSchedule"   containers:   - image: nvidia/cuda:11.0-base     name: mig-none-example     resou...

  The simplest way to achieve this goal is using KubernetesPodOperator to execute papermill command. For example run_notebook = kubernetes_pod_operator.KubernetesPodOperator( task_id=f"run-notebook", name=f"run-notebook", namespace='default', is_delete_operator_pod=True, image_pull_policy="IfNotPresent", startup_timeout_seconds=3600, cmds=['/bin/bash'], arguments=["-c", """ echo y | conda create --name=runenv python=3.8 source /opt/conda/etc/profile.d/conda.sh conda activate runenv conda install ipykernel python -m ipykernel install --user --name=runenv papermill \ "{{dag_run.conf['nbIn']}}" \ "{{dag_run.conf['nbOut']}}" \ ...