Pyspark yarn cluster模式写入HDFS报目录已存在错误是什么原因？

1. Spark内置的saveAsTextFile()接口有强制校验规则：要求传入的输出路径必须不存在，否则会直接抛出FileAlreadyExistsException，避免已有数据被意外覆盖，这是框架的原生设计。
2. 不带--master yarn --deploy-mode cluster参数运行正常的原因：
   缺省配置下Spark会以local[*]本地模式运行，若你的本地环境没有配置集群HDFS的客户端参数，代码中的hdfs:///user/output会被解析为本地文件系统的路径，而非集群HDFS路径，本地对应路径不存在，所以运行成功。
3. 带YARN集群模式参数报错的原因：
   指定YARN集群模式运行时，Spark会读取集群的HDFS配置，hdfs:///user/output会被正确解析为集群HDFS上的路径hdfs://tnhanComputer:9000/user/output，该路径此前已被创建（比如之前测试任务生成后未删除、其他用户创建过该路径），触发框架的路径存在校验，直接报错。
4. 仅计算机A有报错、计算机B运行正常的原因：
   YARN集群模式下，Spark的Driver进程会被YARN随机调度到某个从节点的容器中运行，本次运行Driver正好被调度到计算机A上，所以报错日志输出在A的stdout中；计算机B上仅运行了无异常的Executor进程，因此没有报错日志。

你可以根据需求选择以下任意一种方式解决：

• 每次运行任务前手动删除已存在的HDFS目标路径：执行HDFS命令hdfs dfs -rm -r /user/output
• 在代码中添加保存前删除路径的逻辑，自动清理历史路径：

def save_file_hdfs(data, session, path):
    """
    Use this to save the file to HDFS.
    The saved file will be named "part-00000"
    """
    # 保存前先删除已存在的目标路径
    hadoop_conf = session.sparkContext._jsc.hadoopConfiguration()
    fs = session._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
    fs.delete(session._jvm.org.apache.hadoop.fs.Path(path), True)
    
    rdd_list = session.sparkContext.parallelize(data)
    rdd_list.coalesce(1).map(lambda row: str(row)).saveAsTextFile(path)

• 改用DataFrame写接口，指定覆盖模式自动覆盖已有路径：

def save_file_hdfs(data, session, path):
    rdd_list = session.sparkContext.parallelize(data)
    # 转为DataFrame后指定覆盖模式写入
    session.createDataFrame(rdd_list.map(lambda x: (x,)), schema=["content"])\
           .write.mode("overwrite").option("encoding", "utf-8").text(path)

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