Я определяю следующий код для загрузки предварительно обученной модели встраивания:
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 раза (= количество разделов). Если мой ответ правильный, недостаток такого моделирования связан с размером файла modelpath. Предположим, этот файл имеет размер 10 ГБ и предположим, что у меня 200 разделов. Таким образом, в этом случае нам понадобится 10 * 200 ГБ = 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
В этом случае сколько раз будет загружен файл modelpath? Обратите внимание, что я установил spark.executor.instances" to 3
