PySpark Overwrite 添加了 sc.addPyFile

Question

我在此路径下保存了这两个文件：

C:\code\sample1\main.py

def method():
    return "this is sample method 1"

C:\code\sample2\main.py

def method():
    return "this is sample method 2"

然后我运行这个：

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext()
spark = SparkSession(sc)

sc.addPyFile("~/code/sample1/main.py")
main1 = __import__("main")
print(main1.method()) # this is sample method 1

sc.addPyFile("~/code/sample2/main.py") # Error

错误是

Py4JJavaError：调用 o21.addFile 时出错。: org.apache.spark.SparkException: 文件 C:\Users\hans.yulian\AppData\Local\Temp\spark-5da165cf-410f-4576-8124-0ab23aba6aa3\userFiles-25a7ca23-84fb-42b7-969d67fmain .py 存在并且与 /C:/Users/hans.yulian/Documents/spark-test/main2/main.py 的内容不匹配

这意味着它的临时文件夹中已经有“main.py”文件并且内容不同。我想知道这种情况是否有任何解决方法，但对我来说，我有以下限制：

1. 文件名仍然必须是“main.py”，只有文件夹可以不同
2. 可以以某种方式清除临时文件夹以添加 aga
3. 在另一个文件中我唯一的解决方案是在 main.py 前面附加随机字符串，例如abcdemain.py和 fghijmain.py，然后我将导入 main = __import__("abcdemain")，但这个不是真的可取

Answer 1

虽然在技术上是可行的，但通过设置spark.files.overwrite 为"true"：

from pyspark import SparkConf, SparkContext

sc = SparkContext(conf=SparkConf().set("spark.files.overwrite", "true"))

并且在简单的情况下会给出正确的结果

def f(*_):                                                                   
    from main import method
    return [method()]

sc.addFile("/path/to/sample1/main.py") 
sc.parallelize([], 3).mapPartitions(f).collect()

from pyspark import SparkConf, SparkContext

sc = SparkContext(conf=SparkConf().set("spark.files.overwrite", "true"))

def f(*_):                                                                   
    from main import method
    return [method()]

sc.addFile("/path/to/sample1/main.py") 
sc.parallelize([], 3).mapPartitions(f).collect()

['this is sample method 1',
'this is sample method 1',
'this is sample method 1']

它在实践中并不可靠，即使您reload在每次访问时使用模块，也会使您的应用程序难以推理。由于 Spark 可能会隐式缓存某些对象，或者透明地重新启动 Python 工作程序，因此您很容易陷入不同节点看到不同源状态的情况。