从Airflow提交PySpark任务：Databricks环境配置与LivyOperator优化

一、LivyOperator提交PySpark任务的简洁实现方式 #

除了传入Python文件列表，还有两种更简洁的写法：

• 内联代码提交：直接用python_code参数写入PySpark代码，不用单独维护外部文件，适合逻辑简单的任务。示例：

from airflow.providers.apache.livy.operators.livy import LivyOperator

livy_inline_task = LivyOperator(
    task_id="run_pyspark_inline",
    livy_conn_id="livy_default",
    python_code="""
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv("/path/to/data.csv")
df.show()
""",
    executor_cores=2,
    executor_memory="4g"
)

• 分布式存储脚本引用：把Python脚本上传到HDFS这类分布式存储，用file参数直接指定存储路径，不用管理本地文件列表，降低Airflow侧的维护成本。示例：

livy_hdfs_task = LivyOperator(
    task_id="run_pyspark_hdfs",
    livy_conn_id="livy_default",
    file="hdfs:///path/to/your_script.py",
    args=["param1", "param2"]  # 脚本所需的运行参数
)

二、LivyOperator配置第三方库的方式 #

Livy本身不支持虚拟环境隔离，但可以通过两种方式安装第三方库：

• 任务时动态上传依赖：用py_files参数上传wheel包，或者通过conf参数指定Maven仓库的Java关联依赖（适配PySpark需要的Java库）。示例：

livy_lib_task = LivyOperator(
    task_id="run_pyspark_with_libs",
    livy_conn_id="livy_default",
    file="hdfs:///path/to/script.py",
    py_files=["hdfs:///path/to/pandas-2.1.0-py3-none-any.whl"],
    conf={
        "spark.jars.packages": "com.databricks:spark-xml_2.12:0.15.0"
    }
)

• 集群全局预安装：在所有Spark节点上用pip install全局安装依赖，适合多个任务共用的库，但没有环境隔离性。

三、Airflow提交Databricks PySpark任务的虚拟环境与库配置方法 #

推荐用DatabricksSubmitRunOperator实现，具体配置方式如下：

1. 动态指定任务依赖库 #

直接在算子的libraries参数中声明需要安装的库，支持PyPI包、自定义wheel包、Maven库等类型，提交任务时会自动在集群安装。示例：

from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator

databricks_task = DatabricksSubmitRunOperator(
    task_id="run_databricks_pyspark",
    databricks_conn_id="databricks_default",
    new_cluster={
        "spark_version": "13.3.x-scala2.12",
        "node_type_id": "m5.xlarge",
        "num_workers": 2
    },
    libraries=[
        {"pypi": {"package": "pandas==2.1.0"}},
        {"whl": "dbfs:/path/to/your_custom_lib.whl"}
    ],
    spark_python_task={
        "python_file": "dbfs:/path/to/your_script.py",
        "parameters": ["arg1", "arg2"]
    }
)

2. 配置独立虚拟环境 #

如果需要严格的环境隔离，可通过集群初始化脚本实现：

• 先在Databricks工作区创建conda/venv环境，导出环境配置文件到DBFS；
• 编写初始化脚本加载该环境，示例脚本：

# dbfs:/path/to/init_env.sh
conda env create -f dbfs:/path/to/env.yaml
conda activate your_env_name

• 提交任务时在集群配置中指定初始化脚本：

new_cluster={
    "spark_version": "13.3.x-scala2.12",
    "node_type_id": "m5.xlarge",
    "num_workers": 2,
    "init_scripts": [{"dbfs": {"destination": "dbfs:/path/to/init_env.sh"}}]
}

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