AWS MWAA Airflow自定义.cfg文件放置与读取报错求助

问题分析 #

出现configparser.NoSectionError: No section: 'S3'错误，本质是代码读取dev.cfg时要么文件不存在，要么文件内容未被正确加载，导致ConfigParser无法找到指定section。结合你的场景（无控制台权限、依赖S3同步文件、用插件设置环境变量），核心问题大概率出在路径指向错误、文件未同步到MWAA本地，或者DAG读取配置的逻辑有疏漏。

排查与修复步骤 #

1. 验证配置文件路径与存在性 #

首先确认插件设置的DEV_CONFIG_PATH是否指向正确的本地文件，以及文件是否已从S3同步到MWAA：

• 修改插件代码，添加日志打印（可在CloudWatch的MWAA webserver日志组查看输出）：

import os
import logging

from airflow.plugins_manager import AirflowPlugin

logger = logging.getLogger(__name__)

airflow_home = os.getenv('AIRFLOW_HOME')
dev_config_path = os.path.join(airflow_home, 'dags', 'dev.cfg')
os.environ["DEV_CONFIG_PATH"] = dev_config_path

# 打印关键信息到日志
logger.info(f"DEV_CONFIG_PATH set to: {dev_config_path}")
logger.info(f"Config file exists: {os.path.exists(dev_config_path)}")
if os.path.exists(dev_config_path):
    with open(dev_config_path, 'r') as f:
        logger.info(f"Config content:\n{f.read()}")

class EnvVarPlugin(AirflowPlugin):
    name = 'env_var_plugin'

• 检查日志：
  • 如果路径错误：确认MWAA的AIRFLOW_HOME实际值（默认是/usr/local/airflow），调整路径拼接逻辑。
  • 如果文件不存在：等待S3到MWAA的同步完成（通常1-5分钟），或重新上传dev.cfg触发同步。
  • 如果文件内容为空或不正确：重新上传正确的dev.cfg到S3的dags目录。

2. 修复DAG中的配置读取逻辑 #

确保DAG代码正确使用环境变量，且在读取前验证文件状态：

import configparser
import os

# 从环境变量获取路径，同时设置默认值兜底
dev_config_path = os.environ.get('DEV_CONFIG_PATH', os.path.join('/usr/local/airflow', 'dags', 'dev.cfg'))

# 先验证文件存在
if not os.path.isfile(dev_config_path):
    raise FileNotFoundError(f"Config file missing at: {dev_config_path}")

config = configparser.ConfigParser()
config.read(dev_config_path)

# 读取配置时捕获异常，便于排查
try:
    bucket_name = config.get('S3', 'bucket_name')
except configparser.NoSectionError:
    raise ValueError(f"Config file {dev_config_path} has no [S3] section")
except configparser.NoOptionError:
    raise ValueError(f"Config file {dev_config_path} missing 'bucket_name' in [S3] section")

3. 替代方案：跳过插件直接在DAG中构造路径 #

如果插件加载机制导致环境变量未及时生效，可以直接在DAG中构造配置文件路径，避免依赖插件：

import configparser
import os

# 直接构造路径，无需插件设置环境变量
DEV_CONFIG_PATH = os.path.join(os.getenv('AIRFLOW_HOME', '/usr/local/airflow'), 'dags', 'dev.cfg')

if not os.path.isfile(DEV_CONFIG_PATH):
    raise FileNotFoundError(f"Cannot locate dev.cfg at {DEV_CONFIG_PATH}")

config = configparser.ConfigParser()
config.read(DEV_CONFIG_PATH)
bucket_name = config.get('S3', 'bucket_name')

4. 额外检查点 #

• 确认S3中的dev.cfg内容无格式错误：[S3]标签拼写正确，无多余空格，bucket_name赋值正常。
• 检查MWAA的同步状态：如果多次上传文件仍未同步，可查看MWAA环境的日志确认同步服务是否正常。

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