求助：在Airflow DAG中用Pandas读取S3文件时无限运行无报错

可能的原因及解决方案 #

1. s3fs未正确获取Airflow的AWS凭证 #

虽然你能通过Airflow列出S3桶，但s3fs可能无法自动继承Airflow的AWS连接配置，导致请求挂起。

解决办法： 在代码中显式从Airflow的AWS连接获取凭证，初始化s3fs文件系统后再操作：

from airflow.providers.amazon.aws.hooks.s3 import S3Hook
import pandas as pd
import s3fs

def process_s3_data():
    # 替换为你的Airflow AWS连接ID
    s3_hook = S3Hook(aws_conn_id="aws_default")
    credentials = s3_hook.get_credentials()

    # 初始化带凭证的S3文件系统
    fs = s3fs.S3FileSystem(
        key=credentials.access_key,
        secret=credentials.secret_key,
        token=credentials.token  # 若使用临时凭证需添加
    )

    # 读取S3中的Excel文件
    with fs.open("s3://your-bucket/path/file.xlsx", "rb") as f:
        df = pd.read_excel(f)

    # 数据处理逻辑...

    # 将DataFrame写入S3
    with fs.open("s3://your-bucket/path/output.csv", "wb") as f:
        df.to_csv(f, index=False)

2. Docker容器网络或代理问题 #

Airflow所在的Docker容器可能存在网络阻塞，无法正常连接S3端点，导致请求无限等待。

解决办法：

• 检查容器网络模式：若使用bridge模式，确认容器能访问外部网络（可尝试在容器内执行curl https://s3.amazonaws.com测试）。
• 若部署在VPC内，需配置S3 VPC端点，确保容器能直接访问S3而无需走公网。
• 检查容器是否有代理配置冲突，若有，需在容器环境变量中设置正确的代理参数（HTTP_PROXY/HTTPS_PROXY）。

3. s3fs版本兼容性问题 #

Airflow容器内的s3fs版本与本地环境不一致，可能导致API调用异常。

解决办法：

• 查看本地环境的s3fs版本：pip show s3fs
• 在Airflow的Docker构建文件（如requirements.txt）中指定相同版本：

  s3fs==2023.6.0  # 替换为你的本地版本号
  pandas==2.0.3
  

4. 任务超时设置缺失 #

s3fs默认超时时间过长，导致连接异常时任务无限挂起而非报错。

解决办法： 初始化s3fs时添加超时配置，快速暴露问题：

fs = s3fs.S3FileSystem(
    key=credentials.access_key,
    secret=credentials.secret_key,
    config_kwargs={"connect_timeout": 10, "read_timeout": 30}
)

5. Airflow执行器资源限制 #

若使用CeleryExecutor等分布式执行器，worker节点的CPU/内存不足可能导致任务卡住。

解决办法：

• 查看worker节点的资源使用情况（docker stats）。
• 调整Airflow worker的资源分配（如在docker-compose.yml中增加cpu_shares或mem_limit）。

额外排查步骤 #

• 查看Airflow任务的完整日志（包括worker节点日志），可能存在未显示的权限或连接错误。
• 测试在Airflow容器内直接运行Python代码（docker exec -it <airflow-worker-container> python），模拟DAG中的操作，确认是否能复现问题。

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