时间戳

首页 > 解决方案 > Spark 执行器结果“通过 BlockManager 发送”是什么意思？

docker - Spark 执行器结果“通过 BlockManager 发送”是什么意思？

问题描述

我有一个运行 spark-master 的主机和 3 个 spark-workers，都在 docker 容器中。我有另一台主机充当 Spark 驱动程序，从第一台主机读取数据。

只要返回的数据很小（<6000 行），我就能够成功地从第一台主机检索数据但是当我尝试读取大块（100k+ 行）时它失败了。

我检查了执行程序日志，当读取成功时，我收到以下日志消息：

19/07/23 21:54:17 INFO CassandraConnector: Connected to Cassandra cluster: DataMonitor
19/07/23 21:54:17 INFO Executor: Finished task 0.0 in stage 1.0 (TID 4). 1014673 bytes result sent to driver
19/07/23 21:54:24 INFO CassandraConnector: Disconnected from Cassandra cluster: DataMonitor

但是当读取不成功时，我会收到以下日志消息：

19/07/23 22:21:55 INFO CassandraConnector: Connected to Cassandra cluster: DataMonitor
19/07/23 22:22:03 INFO MemoryStore: Block taskresult_13 stored as bytes in memory (estimated size 119.2 MB, free 2.4 GB)
19/07/23 22:22:03 INFO Executor: Finished task 0.3 in stage 4.0 (TID 13). 124969484 bytes result sent via BlockManager)
19/07/23 22:22:10 INFO CassandraConnector: Disconnected from Cassandra cluster: DataMonitor

看起来当结果足够大时，它会“通过 BlockManager 发送”，但当结果足够小时，它会“发送给驱动程序”。

那么我如何获得它以便将每个结果都发送给驱动程序？

标签： dockerapache-sparkcassandrapysparkcontainers

解决方案

每个 Executor 运行任务并将任务的结果发送回驱动程序。如果任务结果小，则直接与任务状态一起发送，但如果结果大，则按以下公式计算：

taskResultSize > conf.getSizeAsBytes("spark.task.maxDirectResultSize", 1L << 20)

或者

taskResultSize > conf.get("spark.driver.maxResultSize")

源代码

Executor 将结果存储在本地磁盘上并将IndirectTaskResult发送blockId回驱动程序。

然后驱动程序使用 netty viaBlockManager下载远程结果。
看看这里。

如果不够详细，请告诉我。

推荐阅读