如何在PySpark Executor中执行HDFS文件系统操作？

问题背景 #

在Driver端实现ABFSS文件系统操作十分简单，但Executor端因缺少SparkContext、无法直接复用Driver配置等问题，导致并行文件操作难以实现，尝试设置spark.executor.allowSparkContext=true也未解决问题。

Driver端可行代码：

Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem

# 通过spark.conf.set()配置OAuth相关参数
spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
# 其他必要OAuth配置项

fs = Path('<abfss-path>').getFileSystem(sc._jsc.hadoopConfiguration())
data_output_stream = fs.create(Path('<abfss-path>/<file-name>'), True, num_bytes)

Executor端待实现的代码框架：

def do_some_io(i):
    fs = Path('<abfss-path>').getFileSystem(<some-magic-conf-here>)
    data_output_stream = fs.create(Path('<abfss-path>/<file-name>'), True, num_bytes)

rdd = sc.parallelize(range(0, 100))
rdd.foreach(lambda i : do_some_io(i))

解决方案 #

方法1：利用Executor自动同步的Hadoop配置 #

Spark会自动将Driver端的Hadoop配置同步到所有Executor节点，无需手动传递配置。直接在Executor端获取JVM的Hadoop配置对象即可：

def do_some_io(i):
    from pyspark import SparkContext
    # 获取当前Executor的JVM网关
    gateway = SparkContext._active_spark_context._gateway
    Path = gateway.jvm.org.apache.hadoop.fs.Path
    # 获取Executor端的HadoopConfiguration（已同步Driver配置）
    hadoop_conf = gateway.jvm.org.apache.hadoop.conf.Configuration()
    
    fs = Path('<abfss-path>').getFileSystem(hadoop_conf)
    # 生成唯一文件名，避免多Executor写入冲突
    unique_file_path = Path(f'<abfss-path>/task_{i}_output.txt')
    data_output_stream = fs.create(unique_file_path, True, 4096)
    # 写入示例内容
    data_output_stream.write(f"Executor任务{i}输出内容".encode('utf-8'))
    # 必须关闭流，确保数据写入完成
    data_output_stream.close()

rdd = sc.parallelize(range(0, 100))
rdd.foreach(do_some_io)

方法2：广播关键配置项（针对特殊场景） #

若部分配置未自动同步，可在Driver端收集关键配置并广播到Executor：

# Driver端收集需要的ABFSS配置项
required_conf = {
    "fs.azure.account.auth.type": spark.conf.get("fs.azure.account.auth.type"),
    "fs.azure.account.oauth.provider.type": spark.conf.get("fs.azure.account.oauth.provider.type"),
    "fs.azure.account.oauth2.client.id": spark.conf.get("fs.azure.account.oauth2.client.id"),
    "fs.azure.account.oauth2.client.secret": spark.conf.get("fs.azure.account.oauth2.client.secret"),
    "fs.azure.account.oauth2.endpoint": spark.conf.get("fs.azure.account.oauth2.endpoint")
}
# 广播配置到所有Executor
broadcast_conf = sc.broadcast(required_conf)

def do_some_io(i):
    from pyspark import SparkContext
    gateway = SparkContext._active_spark_context._gateway
    Path = gateway.jvm.org.apache.hadoop.fs.Path
    hadoop_conf = gateway.jvm.org.apache.hadoop.conf.Configuration()
    
    # 将广播的配置项注入到Hadoop配置中
    for key, value in broadcast_conf.value.items():
        hadoop_conf.set(key, value)
    
    fs = Path('<abfss-path>').getFileSystem(hadoop_conf)
    unique_file_path = Path(f'<abfss-path>/task_{i}_output.txt')
    data_output_stream = fs.create(unique_file_path, True, 4096)
    data_output_stream.write(f"Executor任务{i}输出内容".encode('utf-8'))
    data_output_stream.close()

rdd = sc.parallelize(range(0, 100))
rdd.foreach(do_some_io)

核心注意事项 #

• 避免文件冲突：多个Executor同时写入同一文件会导致数据损坏，必须为每个任务生成唯一文件名（如用任务索引、ID命名）
• 资源释放：务必调用close()关闭输出流，否则可能出现数据未完全写入的情况
• 权限验证：确保Executor节点拥有ABFSS路径的访问权限，OAuth配置需完整同步

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