scala - 如何在 Spark 中对每个执行器执行一次操作

Question

我有一个存储在 S3 中的 weka 模型，其大小约为 400MB。
现在，我有一组记录，我想在这些记录上运行模型并执行预测。

为了执行预测，我尝试过的是，

在驱动程序上下载并加载模型作为静态对象，将其广播给所有执行程序。对预测 RDD 执行映射操作。
----> 不工作，如在 Weka 中执行预测，模型对象需要修改并且广播需要只读副本。

将模型作为静态对象下载并加载到驱动程序上，并在每个映射操作中将其发送到执行程序。
-----> 工作(效率不高，因为在每个 map 操作中，我传递了 400MB 对象)

在驱动程序上下载模型并将其加载到每个执行器上并将其缓存在那里。 (不知道怎么做)

有人知道如何在每个执行器上加载模型一次并缓存它，以便我不会再次加载其他记录吗？

Answer 1

您有两个选择:

1. 创建一个用惰性 val 表示数据的单例对象:

    object WekaModel {
        lazy val data = {
            // initialize data here. This will only happen once per JVM process
        }
    }       

然后，您可以在 map 中使用惰性 val。功能。 lazy val确保每个工作 JVM 初始化他们自己的数据实例。不会为 data 执行序列化或广播.

    elementsRDD.map { element =>
        // use WekaModel.data here
    }

优势

效率更高，因为它允许您为每个 JVM 实例初始化一次数据。例如，当需要初始化数据库连接池时，这种方法是一个不错的选择。

缺点

对初始化的控制较少。例如，如果您需要运行时参数，则初始化您的对象会比较棘手。

如果需要，您无法真正释放或释放对象。通常，这是可以接受的，因为操作系统会在进程退出时释放资源。

2. 使用 mapPartition (或 foreachPartition )方法在 RDD 上，而不仅仅是 map .

这允许您初始化整个分区所需的任何内容。

    elementsRDD.mapPartition { elements =>
        val model = new WekaModel()

        elements.map { element =>
            // use model and element. there is a single instance of model per partition.
        }
    }

优势 :

在对象的初始化和取消初始化方面提供更大的灵活性。

缺点

每个分区都会创建并初始化对象的一个​​新实例。根据每个 JVM 实例有多少个分区，这可能是也可能不是问题。