pySpark addfile 选项，执行程序中的工作人员会发生什么答案

【问题标题】：pySpark addfile option, what happens on worker in executorpySpark addfile 选项，执行程序中的工作人员会发生什么
【发布时间】：2023-03-20 17:05:01
【问题描述】：

很明显，为了分发小的查找数据，最好使用广播变量。

假设我们以纱线客户端模式从主节点运行 pySpark 代码（火花提交）。因此应用程序驱动程序将始终在主节点上创建。我们从主节点上的本地路径读取文件。

with open('/tmp/myfile.txt', 'r') as f:
    lookup = {}
    for line in f.readlines():
        line = parse(line) # Method parse uses re and return dict
        lookup[line['name']] = line['age']

然后我们创建broadcast var并使用它：

lookupBC = sc.broadcast(lookup)

output = sc.textFile('/path/to/hdfs/')\
    .map(lambda e: (lookupBC.value.get(e, e), 1))\
    .collect()

在我们的例子中，这个 bc var 是在驱动程序（主节点）上创建的，并且 spark 在集群中的所有数据节点之间复制这个 var，其中创建了执行程序，将其保存在这些节点的内存中。因此文件将被读取一次，然后分发给执行者。

如果我们使用 addFile 选项会发生什么？

sc.addFile('/tmp/myfile.txt')

with open(SparkFiles.get('/tmp/myfile.txt')) as f:
    lookup = {}
    for line in f.readlines():
        line = parse(line) # Method parse uses re and return dict
        lookup[line['name']] = line['age']

output = sc.textFile('/path/to/hdfs/')\
    .map(lambda e: (lookup.get(e, e), 1))\
    .collect()

Spark 会将文件'/tmp/myfile.txt' 复制到将创建执行程序的每个节点。那么：

文件将被读取多少次？每个执行者在特定节点上执行一次？还是每个任务一次？
有哪些步骤，如何在 executor 上处理代码？
addFile 或 bc var 用什么更好？
spark 会根据 pyspark 代码做任何优化并创建隐式 bc vars 吗？

在执行程序日志中，我看到了有关 bc vars 的信息，但我的代码中没有使用任何信息：

18/03/21 15:36:27 INFO util.Utils: Fetching spark://172.25.235.201:36478/files/myfile.txt to /data/disk01/yarn/nm/usercache/testuser/appcache/application_1520754626920_6227/spark-f3d19076-0642-4db8-961d-99daae0dfaff/fetchFileTemp230224632617642846.tmp
18/03/21 15:36:27 INFO util.Utils: Copying /data/disk01/yarn/nm/usercache/testuser/appcache/application_1520754626920_6227/spark-f3d19076-0642-4db8-961d-99daae0dfaff/-17884647971521635771454_cache to /data/disk01/yarn/nm/usercache/testuser/appcache/application_1520754626920_6227/container_1520754626920_6227_01_000002/./myfile.txt
18/03/21 15:36:28 INFO broadcast.TorrentBroadcast: Started reading broadcast variable 1
18/03/21 15:36:28 INFO client.TransportClientFactory: Successfully created connection to strt01we.ebb.er.com/172.25.235.216:43791 after 4 ms (0 ms spent in bootstraps)
18/03/21 15:36:28 INFO memory.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 6.3 KB, free 366.3 MB)
18/03/21 15:36:28 INFO broadcast.TorrentBroadcast: Reading broadcast variable 1 took 551 ms

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

广播变量似乎被加载到内存中，直到它们被显式销毁。相比之下，sc.addFile 似乎正在创建一个到磁盘的副本（对于每个执行程序）。所以我猜SparkFiles.get() 会将文件加载到内存中，每次都会被调用。

因此，在上面的示例中，它将被加载一次。
但是，如果您在 .map() 中调用 SparkFiles.get()，它会尝试为 RDD 中的每个条目重新加载文件。

最后，回答你的问题，

文件将被读取多少次？每个执行者在特定节点上执行一次？还是每个任务一次？

取决于，.get 被调用的位置，如上所述。

会有哪些步骤，代码将如何在 executor 上处理？

这部分我没看懂。

addFile 或 bc var 用什么更好？

这些是不同的用例。例如，考虑我们有一个 1GB sqliteDB 转储的情况。 Spark 可以通过 JDBC 连接到这个 DB 对象。它并不真的需要将整个对象加载到内存中。

spark 会基于 pyspark 代码进行任何优化并创建隐式 bc vars 吗？

不确定，但我不这么认为。

【讨论】：