使用Airflow和Docker部署Flask服务后连接失败求助

核心问题分析 #

1. Docker端口未映射：当前docker-compose.yml仅暴露Airflow Web UI的8080端口，Flask服务的5000端口未对外映射，导致外部无法访问。
2. Airflow任务设计冲突：Airflow是短周期批处理任务调度工具，而你的ml-pipeline.py启动的是常驻Flask服务，会导致Airflow任务一直处于"运行中"状态，不符合Airflow的使用场景。
3. 网络地址不匹配：推理脚本中使用的172.21.0.1可能并非容器的正确可访问地址，且错误日志显示实际请求的是localhost，存在地址不一致问题。

分步解决方案 #

1. 修正Docker端口映射 #

修改docker-compose.yml，添加5000端口的映射配置，让外部可以访问容器内的Flask服务：

version: '3'

services:
  sleek-airflow:
    image: pythonairflow:latest
    volumes:
      - ./airflow:/opt/airflow
    ports:
      - "8080:8080"
      - "5000:5000"  # 添加Flask端口映射
    command: airflow standalone

2. 调整Flask服务的启动方式 #

方案A：容器内同时启动Airflow和Flask（快速适配）

修改docker-compose.yml的command，让Airflow和Flask同时后台运行：

command: >
  bash -c "airflow standalone & python /ml-pipeline.py"

这样两个服务会同时在容器内运行，不会互相阻塞。

方案B：拆分模型训练与服务部署（推荐）

Airflow只负责模型训练，Flask服务单独部署，更符合工具定位：

• 拆分出模型训练脚本train_model.py：

from sklearn.datasets import fetch_california_housing
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
import pickle

# 训练流程
data = fetch_california_housing()
X = data.data
y = data.target

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

model = LinearRegression()
model.fit(X_scaled, y)

# 保存模型和scaler到共享目录
with open('/opt/airflow/model.pkl', 'wb') as f:
    pickle.dump(model, f)
with open('/opt/airflow/scaler.pkl', 'wb') as f:
    pickle.dump(scaler, f)

• 拆分出Flask服务脚本predict_service.py：

from flask import Flask, request, jsonify
import pickle

app = Flask(__name__)

# 加载预训练的模型和scaler
with open('/opt/airflow/model.pkl', 'rb') as f:
    model = pickle.load(f)
with open('/opt/airflow/scaler.pkl', 'rb') as f:
    scaler = pickle.load(f)

@app.route('/predict', methods=['POST'])
def predict():
    try:
        data = request.get_json(force=True)
        features = data['features']
        scaled_features = scaler.transform([features])
        prediction = model.predict(scaled_features)[0]
        return jsonify({'prediction': prediction})
    except Exception as e:
        return jsonify({'error': str(e)})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

• 修改DAG仅执行训练任务：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

dag = DAG(
    'pipeline_dag',
    default_args={'start_date': days_ago(1)},
    schedule_interval='0 23 * * *',
    catchup=False
)

train_model_task = BashOperator(
    task_id='train_model',
    bash_command='python /train_model.py',
    dag=dag,
)

train_model_task

3. 修正推理脚本的访问地址 #

根据运行环境调整地址：

• 主机上运行脚本：使用http://localhost:5000/predict
• 其他Docker容器内运行：使用容器名称sleek-airflow作为地址：http://sleek-airflow:5000/predict

修正后的推理脚本：

import requests

url = 'http://localhost:5000/predict'
data = {'features': [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8]}

try:
    response = requests.post(url, json=data)
    response.raise_for_status()
    prediction = response.json()['prediction']
    print(f'Predicted Value: {prediction}')
except requests.exceptions.RequestException as e:
    print(f'Request failed: {e}')

验证步骤 #

1. 重新构建镜像：docker build -t pythonairflow:latest .
2. 启动容器：docker-compose up -d
3. 等待服务启动完成后，运行推理脚本验证结果

内容的提问来源于stack exchange，提问作者rajtilakjee