Airflow DAG中创建SparkSession报Java gateway进程退出如何处理

Java gateway process exited before sending its port number报错的核心触发逻辑有两个：

1. DAG解析阶段（Airflow scheduler/worker/webserver加载DAG文件时）就执行了SparkSession初始化代码，此时还没进入任务实际运行的上下文
2. Airflow运行环境缺少Spark启动必需的Java、Spark客户端依赖或对应环境变量配置

1. 基础环境依赖配置 #

所有会解析DAG、运行任务的Airflow节点（scheduler、worker、webserver）必须提前配置好以下环境，且Airflow运行用户有权限访问：

• 安装与PySpark大版本完全匹配的JDK（比如Spark3.x对应JDK8/11），正确配置JAVA_HOME环境变量，执行java -version可正常返回版本信息
• 部署与PySpark版本完全一致的Spark客户端，正确配置SPARK_HOME环境变量，执行spark-submit --version可正常返回版本信息
• 配置PYSPARK_PYTHON环境变量指向任务运行用的Python解释器路径，避免Python版本不匹配问题

注意：不要只在当前用户的.bashrc/.zshrc里配置这些环境变量，Airflow的后台服务不会自动加载shell配置，需要把环境变量写到Airflow服务的启动配置里（比如systemd配置、docker-compose环境变量配置、airflow.cfg的env配置段）。

2. 修正SparkSession初始化位置 #

绝对不要在DAG文件的全局顶层作用域写SparkSession初始化代码，Airflow加载DAG时会直接执行所有顶层代码，此时就会尝试拉起Spark Java进程，直接触发报错。
正确的做法是把SparkSession初始化逻辑放到任务的执行函数内部，只有任务被实际调度运行时才会执行初始化。
错误写法（即触发本次报错的写法）：

# 顶层全局作用域，DAG加载就执行
spark = SparkSession.builder.getOrCreate()

with DAG(...) as dag:
    # 任务逻辑

正确写法示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from pyspark.sql import SparkSession
from datetime import datetime

def spark_job():
    # 初始化逻辑放在任务函数内部，任务运行时才执行
    spark = SparkSession.builder \
        .master("yarn") # 替换成实际的Spark集群master地址
        .appName("airflow_spark_job") \
        .getOrCreate()
    
    # 编写Spark业务逻辑
    test_df = spark.createDataFrame([(1, "test")], ["id", "content"])
    test_df.show()
    
    spark.stop()

with DAG(
    dag_id="wip_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    run_spark_task = PythonOperator(
        task_id="run_spark",
        python_callable=spark_job
    )

3. 生产环境优先用官方提交算子 #

生产场景不推荐在PythonOperator内直接初始化SparkSession，这种方式是在Airflow worker本地拉起Spark进程，资源隔离差、稳定性低。优先使用Airflow官方提供的Spark提交算子，直接将任务提交到Spark集群运行，不需要在Airflow节点本地启动Spark gateway进程，从根源上避免这类DAG加载报错，只需要提前在Airflow连接配置中填写Spark集群地址、依赖包路径即可。

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