如何在PySpark中读取S3文件时忽略不存在的路径

配置未生效的原因 #

spark.sql.files.ignoreMissingFiles参数的作用是处理读取任务执行过程中，已被Spark纳入读取列表的文件突然丢失的场景，不支持过滤传入的路径本身不存在的情况，所以设置后不生效。

方案1：自动忽略不存在路径（最简方案，满足需求a） #

直接开启专门用于过滤不存在路径的配置spark.sql.files.ignoreMissingPaths即可，无需修改现有路径拼接逻辑（Spark 2.1+版本支持）：

# 代码内动态配置
spark.conf.set("spark.sql.files.ignoreMissingPaths", True)

# 原有读取逻辑不变
sqlContext.read.parquet(*files_to_read)

也可以在提交Spark作业时通过启动参数全局配置：
spark-submit --conf spark.sql.files.ignoreMissingPaths=true 你的作业脚本.py

方案2：提前校验路径有效性（满足需求b） #

如果需要提前筛选出所有合法路径，或者使用的Spark版本低于2.1没有上述配置，可以通过Hadoop FileSystem接口批量校验路径：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
hadoop_conf = sc._jsc.hadoopConfiguration()
# 初始化FileSystem，自动适配s3/s3a/hdfs等协议
fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)

valid_paths = []
for path_str in files_to_read:
    path_obj = sc._jvm.org.apache.hadoop.fs.Path(path_str)
    # 校验路径是否存在/是否有匹配通配符的文件
    if fs.globStatus(path_obj):
        valid_paths.append(path_str)

# 仅读取合法路径
sqlContext.read.parquet(*valid_paths)

注意：如果集群S3协议默认用s3a，可以把路径前缀的s3://替换为s3a://避免识别异常。

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