Spark独立模式为何尝试在Worker节点加载CSV文件？

在配置了单个Worker节点的Apache Spark独立模式下执行以下PySpark代码，通过spark-submit提交任务时出现文件找不到的错误，但该CSV文件确实存在于Master节点的指定路径中。疑惑为何Spark会尝试在Worker节点加载文件，寻求原因及解决方法。

执行代码 #

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
sc = SparkSession.builder.appName("YourAppName").getOrCreate()

csvFile = 'file:///home/bennison/Documents/yavar-learnings/spark/pyspark/departuredelays.csv'
file_path_on_master = csvFile
# spark.addFile(csvFile)

data_rdd = sc.read.format('csv').option(
    'inferSchema', 'true').option('header', 'true').load(csvFile)
    

data_rdd.show()
data_rdd.createOrReplaceTempView('us_delay_flights_tbl')

# 查询距离超过1000英里的航班
sc.sql('SELECT * FROM us_delay_flights_tbl WHERE distance > 1000 ORDER BY distance DESC').show()

# 查询旧金山(SFO)到芝加哥(ORD)且延误至少2小时的航班
sc.sql('SELECT date, distance, origin, destination FROM us_delay_flights_tbl WHERE origin = "SFO" AND destination = "ORD" AND delay >= 120 ORDER BY delay DESC').show()

# 为所有航班添加延误类型标签
sc.sql(
    'SELECT date, distance, origin, destination, CASE WHEN delay = 0 THEN "No Dely" WHEN delay > 0 AND delay < 2 THEN "Tolerable Delays" WHEN delay > 2 AND delay < 6 THEN "Long Delays" WHEN delay > 6 THEN "Very Long Delays" ELSE "Early" END AS delay_def FROM us_delay_flights_tbl ORDER BY delay DESC').show()

# DataFrame API实现：查询距离超过1000英里的航班
(data_rdd.select("distance", "origin", "destination").where(
    'distance > 1000').orderBy(desc('distance'))).show()

# DataFrame API实现：查询SFO到ORD且延误至少2小时的航班
(data_rdd.select('date', 'distance', 'origin', 'destination').filter((col('origin') == "SFO")
 & (col("destination") == "ORD") & (col("delay") >= 120)).orderBy(desc('delay')).show())

# DataFrame API实现：添加延误类型标签
(data_rdd.select('date', 'distance', 'origin', 'destination', when(col('delay') == 0, 'No Delay').when((col('delay') > 0) & (col('delay') < 2),
 "Tolerable Delays").when((col('delay') > 2) & (col('delay') < 6), "Long Delays").when((col('delay') > 6), "Very Long Delay").otherwise('Early').alias('delay_def')).orderBy(desc(col='delay')).show())

sc.stop()

报错信息 #

Caused by: java.io.FileNotFoundException: 
File file:/home/bennison/Documents/yavar-learnings/spark/pyspark/departuredelays.csv does not exist

It is possible the underlying files have been updated. You can explicitly invalidate
the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by
recreating the Dataset/DataFrame involved.

Spark的文件读取逻辑中，当使用file://协议指定路径时，Spark默认认为该路径在集群所有节点（包括Worker）的本地文件系统上都存在。在独立模式下，Driver运行在Master节点，但实际执行数据读取任务的是Worker节点上的Executor进程——这些Executor会直接从自身节点的本地路径加载文件分片，而非由Master读取后分发数据。你的CSV文件仅存在于Master节点，因此Worker节点的Executor找不到文件，抛出FileNotFoundException。

提供三种可行方案，按需选择：

1. 复制文件到Worker节点的相同路径 #

由于只有单个Worker节点，直接将Master节点上的CSV文件复制到Worker节点的/home/bennison/Documents/yavar-learnings/spark/pyspark/路径下即可，Worker节点的Executor就能找到对应文件。

2. 使用共享文件系统 #

将CSV文件上传到集群可访问的共享存储（如HDFS、NFS等），修改代码中的文件路径为共享存储的协议路径。例如使用HDFS时，路径改为hdfs://<namenode-host>:<port>/path/to/departuredelays.csv，Spark会自动从共享存储读取数据并分发给Worker节点。

3. 使用Spark的文件分发机制（推荐） #

利用Spark的addFile方法将文件自动分发到所有Worker节点的临时目录，修改代码如下：

from pyspark import SparkContext, SparkFiles
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.appName("YourAppName").getOrCreate()

csvFile = '/home/bennison/Documents/yavar-learnings/spark/pyspark/departuredelays.csv'
# 调用addFile分发文件，无需file://协议
spark.sparkContext.addFile(csvFile)

# 通过SparkFiles.get获取Worker节点上的文件路径
data_rdd = spark.read.format('csv').option(
    'inferSchema', 'true').option('header', 'true').load(f"file://{SparkFiles.get('departuredelays.csv')}")

# 后续代码保持不变...

addFile会将Master节点上的文件上传到Spark分布式缓存，自动同步到所有Worker节点，Executor可通过SparkFiles.get获取文件在Worker节点的本地路径。

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