## 使用 Airflow + Docker 在遠端主機部署與執行 DAG
###### tags: `Data Engineering`
###### 更新日期: `2025-10-04`
---
## 需求描述
用 airflow 執行程式

(當 airflow 檔案放入
scheduler and webserver container 的/opt/airflow/dags/資料夾後，
執行airflow dags reserialize 即可重新 detect 新的 dag)

Airflow UI url: (Airflow 遠端主機 IP):8080

---

## 一、SSH 連線到有 Airflow 的遠端主機
ssh (使用者名稱)@(Airflow 遠端主機 IP)

## 二、建立 dags 資料夾
cd /opt/airflow
sudo mkdir dags

## 三、上傳 code

### 1. 因為權限問題, 先上傳 code 到遠端主機的個人資料夾
scp (DAG 檔名).py (使用者名稱)@(Airflow 遠端主機 IP):/home/(使用者名稱)/test/

### 2. 在遠端機器上執行 code 搬移
ssh (使用者名稱)@(Airflow 遠端主機 IP)

sudo mv /home/(使用者名稱)/test/(DAG 檔名).py /opt/airflow/dags/

cd /opt/airflow/dags/

ls

## 四、從主機把 DAG 複製進容器
sudo docker cp /opt/airflow/dags/(DAG 檔名).py \
docker_airflow-scheduler_1:/opt/airflow/dags/

sudo docker cp /opt/airflow/dags/(DAG 檔名).py \
(airflow-webserver 容器名稱):/opt/airflow/dags/

## 五、序列化

### 1. 先停掉 webserver，避免併發寫入
sudo docker stop (airflow-webserver 容器名稱)

### 2. 清空 serialized_dag
sudo docker exec (airflow-scheduler 容器名稱) python - <<'PY'
from airflow.utils.session import create_session
from airflow.models.serialized_dag import SerializedDagModel
with create_session() as s:
n = s.query(SerializedDagModel).delete()
s.commit()
print(f"deleted {n} rows from serialized_dag")
PY

### 3. 只序列化這次要執行的 DAG（避免撞名）
sudo docker exec (airflow-scheduler 容器名稱) airflow dags reserialize --subdir /opt/airflow/dags/(DAG 檔名).py

> 啟動 webserver：

sudo docker start (airflow-webserver 容器名稱)

sudo docker exec (airflow-webserver 容器名稱) airflow dags reserialize --subdir /opt/airflow/dags/(DAG 檔名).py

## 六、執行 DAG (以手動執行為例)
### Airflow UI url: (Airflow 遠端主機 IP):8080

## 七、在 Airflow UI 上查看 DAG 執行狀態

### 1. 在 DAGs 列表中查詢並點選指定的 RAG

### 2. 點選藍色按鈕 Unpause 並進入 RAG 再按下黃圈處的 Run
* 

* 

### 3. Run 完畢後 => 紅色代表有錯誤，綠燈代表正常

### 4. 錯誤排除 (若有錯誤)
依序點選 Graph => 紅色 failed 實體 => 右邊選單會出現 log
點選 log，可以細看 error message (後續再找原因修 bug)