pySpark addfile选项，执行器中的worker会发生什么

Question

很明显，对于分发小的查找数据，使用广播变量更好。

假设我们在yarn客户端模式下从Master节点运行pySpark代码（spark submit）。因此应用程序驱动程序将始终在主节点上创建。我们从主节点上的本地路径读取文件。

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()

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()

1. 文件将被读取多少次？每个执行器在特定节点上执行一次？还是每个任务一次？
2. 步骤是什么，如何在Executor上处理代码？
3. 使用更好的addFile或bc var？
4. spark是否会基于pyspark代码进行任何优化并创建隐式bc变量？

在executor日志中，我看到了有关bc VAR的信息，但我在代码中没有使用任何信息：

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

Answer 1

广播变量似乎被加载在内存中，直到它们被显式销毁。相反，sc.addfile似乎是（为每个执行程序）创建一个磁盘副本。所以我猜测sparkfiles.get()会在每次调用文件时将文件加载到内存中。

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

最后，回答大家的问题，

视情况而定，调用.get的位置，如上所述。

步骤是什么，如何在Executor上处理代码？

我不明白这部分。