如何在每个执行程序中加载一次文件？答案

【问题标题】：How to load a file in each executor once?如何在每个执行程序中加载一次文件？
【发布时间】：2019-02-05 18:37:56
【问题描述】：

我定义了以下代码以加载预训练的嵌入模型：

import gensim

from gensim.models.fasttext import FastText as FT_gensim
import numpy as np

class Loader(object):
    cache = {}
    emb_dic = {}
    count = 0
    def __init__(self, filename):
        print("|-------------------------------------|")
        print ("Welcome to Loader class in python")
        print("|-------------------------------------|")
        self.fn = filename

    @property
    def fasttext(self):
        if Loader.count == 1:
                print("already loaded")
        if self.fn not in Loader.cache:
            Loader.cache[self.fn] =  FT_gensim.load_fasttext_format(self.fn)
            Loader.count = Loader.count + 1
        return Loader.cache[self.fn]


    def map(self, word):
        if word not in self.fasttext:
            Loader.emb_dic[word] = np.random.uniform(low = 0.0, high = 1.0, size = 300)
            return Loader.emb_dic[word]
        return self.fasttext[word]

我把这个类称为：

inputRaw = sc.textFile(inputFile, 3).map(lambda line: (line.split("\t")[0], line.split("\t")[1])).map(Loader(modelpath).map)

我对模型路径文件将被加载多少次感到困惑？我想为每个执行程序加载一次并被其所有核心使用。我对这个问题的回答是模型路径将被加载 3 次（= 分区数。）。如果我的回答是正确的，那么这种建模的缺点与文件模型路径的大小有关。假设这个文件是 10 GB，假设我有 200 个分区。因此，在这种情况下，我们将需要 10*200gb = 2000 且非常大（此解决方案仅适用于少量分区。）

假设我有一个 rdd =(id, sentence) =[(id1, u'patina californian'), (id2, u'virgil american'), (id3', u'frensh'), (id4, u'american')]

我想总结每个句子的嵌入词向量：

def test(document):
    print("document is = {}".format(document))
    documentWords = document.split(" ")
    features = np.zeros(300)
    for word in documentWords:
        features = np.add(features, Loader(modelpath).fasttext[word])
    return features

def calltest(inputRawSource):

    my_rdd = inputRawSource.map(lambda line: (line[0], test(line[1]))).cache()
    return my_rdd

在这种情况下，模型路径文件将被加载多少次？注意我设置了spark.executor.instances" to 3

【问题讨论】：

我认为这行不通而且效率不高。当您调用 Loader(modelpath).map 时，这意味着将所有数据从 spark 发送到 python。我将尝试使用 spark 数据框模型 + UDF 。您可以以 UTF 方式编写您的加载程序。让 spark 处理您的数据分发。
@bib 你可以先看到这个changhsinlee.com/pyspark-udf
ok~ 现在我明白你的问题是你使用python的模型，模型太大了。我徘徊你可以使用模型文件到火花数据框来计算而不加载到gensim.models.KeyedVectors.load_word2vec_format
这与您的问题相似吗？ towardsdatascience.com/…
我知道你的问题，不知道怎么按你的方法解决，比如将模型数据全部加载到内存中（Loader.cache[self.fn] = FT_gensim.load_fasttext_format(self .fn))。我建议你将模型数据加载到数据框中，并尝试通过这种方式做任何你想做的事情。

标签： apache-spark pyspark fasttext

【解决方案1】：

默认情况下，分区数设置为 Spark 集群中所有执行器节点上的内核总数。假设您在包含总共 200 个 CPU 内核的 Spark 集群（或超级计算执行器）上处理 10 GB，这意味着默认情况下，Spark 可能使用 200 个分区来处理您的数据。

此外，为了让所有 CPU 内核在每个执行器上工作，这可以在 python 中解决（使用 100% 的所有内核和多处理模块）。

【讨论】：