Airflow DAG导入失败及触发报错问题求助（Py4JError与BrokenPipeError）

看起来你遇到了Airflow DAG导入和触发时的两类棘手问题，结合你说代码在PyCharm能正常运行的情况，咱们一步步拆解分析：

一、DAG导入时的Py4JError（Parquet相关） #

这个错误提示Py4JError: An error occurred while calling o28.parquet，明显和PySpark操作Parquet文件有关。之所以PyCharm正常但Airflow报错，大概率是环境差异或者代码执行时机的问题：

• 环境依赖不一致：检查Airflow使用的虚拟环境（/Users/soyuz/airflow/venv）是否安装了正确版本的PySpark和Parquet相关依赖（比如pyarrow）。PyCharm的开发环境可能已经配齐了这些，但Airflow的venv里可能缺失或者版本不兼容。你可以激活Airflow的虚拟环境后，运行pip list对比PyCharm环境的依赖列表。
• 代码执行时机错误：Airflow在扫描DAG文件时，会执行所有顶层代码（包括你导入的lib模块里的代码）。如果你的某个lib模块（比如处理Parquet的convert_raw_to_formatted）在导入阶段就初始化了SparkSession或者执行了Parquet读写操作，那Airflow的Scheduler进程在解析DAG时就会触发这个错误——因为Scheduler后台环境可能没有正确配置Spark上下文。
  👉 解决思路：把所有Spark相关的初始化和操作逻辑，移到PythonOperator的python_callable函数内部，不要放在模块的顶层（也就是import时就会跑的代码段）。比如把SparkSession.builder.getOrCreate()这类代码放到convert_raw_to_formatted函数里，而不是模块开头。

二、触发DAG时的BrokenPipeError #

这个错误属于Flask（Airflow Webserver的底层框架）的管道断开问题，一般不是DAG业务代码的直接问题，但很可能是前面的DAG解析异常导致Webserver处理请求时崩溃：

• 重启Airflow进程：先尝试重启Webserver和Scheduler，执行以下命令：

  airflow webserver restart
  airflow scheduler restart
  

  有时候进程卡死或者状态异常会引发这类管道错误，重启后大概率能缓解。
• 查看Webserver日志：去Airflow的默认日志目录~/airflow/logs里找Webserver相关的日志文件，看看有没有更详细的错误堆栈，确认是不是DAG解析失败导致Webserver处理请求时出错。

针对你的DAG代码的额外建议 #

从你贴的DAG代码来看，你导入了大量lib下的模块，务必检查这些模块：

1. 有没有在模块顶层（比如文件开头）就执行Spark操作或者文件读写？如果有，全部移到对应的PythonOperator函数内部。
2. 单独在Airflow的虚拟环境里测试每个python_callable函数，比如激活venv后运行：

   from lib.raw_to_fmt_sirene import convert_raw_to_formatted
   convert_raw_to_formatted(task_number='test')
   

   这样能快速定位到底是函数本身的问题，还是Airflow环境的问题。

另外，确认你的Airflow环境能正确读取Spark的配置：比如SPARK_HOME环境变量是否设置，Scheduler和Worker进程能不能获取到这个变量——如果是本地Spark，这个配置很关键。

备注：内容来源于stack exchange，提问作者stehu