使用Word2VecModel.transform（）在地图功能中不起作用

Question

我已经使用Spark建立了Word2Vec模型并将其另存为模型。现在，我想在另一个代码中将其用作脱机模型。我已经加载了模型，并用它来表示单词的向量（例如，Hello），并且效果很好。但是，我需要使用map在RDD中为许多单词调用它。

当我在地图函数中调用model.transform（）时，它将引发以下错误：

“似乎您正在尝试从广播引用SparkContext。”异常：您似乎正在尝试从广播变量，动作或转换引用SparkContext。SparkContext只能在驱动程序上使用，而不能在工作程序上运行的代码中使用。有关更多信息，请参见SPARK-5063。

编码：

from pyspark import SparkContext
from pyspark.mllib.feature import Word2Vec
from pyspark.mllib.feature import Word2VecModel

sc = SparkContext('local[4]',appName='Word2Vec')

model=Word2VecModel.load(sc, "word2vecModel")

x= model.transform("Hello")
print(x[0]) # it works fine and returns [0.234, 0.800,....]

y=sc.parallelize([['Hello'],['test']])
y.map(lambda w: model.transform(w[0])).collect() #it throws the error

非常感谢您的帮助。

Answer 1

这是预期的行为。像其他MLlib模型一样，Python对象只是Scala模型的包装，实际处理被委派给JVM对应对象。由于工作人员无法访问Py4J网关，因此无法从操作或转换中调用Java /Scala方法。

通常，MLlib模型提供一种可以直接在RDD上运行的辅助方法，但实际情况并非如此。Word2VecModel提供了getVectors一种方法，该方法可将单词从单词返回到向量，但不幸的是它是一个JavaMap这样的方法，因此在转换中不起作用。您可以尝试这样的事情：

from pyspark.mllib.linalg import DenseVector

vectors_ = model.getVectors() # py4j.java_collections.JavaMap
vectors = {k: DenseVector([x for x in vectors_.get(k)])
    for k in vectors_.keys()}

获取Python字典，但速度会非常慢。另一个选择是将该对象以Python可以使用的形式转储到磁盘上，但是需要对Py4J进行一些修改，最好避免这种情况。相反，让我们将模型读取为DataFrame：

lookup = sqlContext.read.parquet("path_to_word2vec_model/data").alias("lookup")

我们将得到以下结构：

lookup.printSchema()
## root
## |-- word: string (nullable = true)
## |-- vector: array (nullable = true)
## |    |-- element: float (containsNull = true)

可以通过以下方式将单词映射到向量join：

from pyspark.sql.functions import col

words = sc.parallelize([('hello', ), ('test', )]).toDF(["word"]).alias("words")

words.join(lookup, col("words.word") == col("lookup.word"))

## +-----+-----+--------------------+
## | word| word|              vector|
## +-----+-----+--------------------+
## |hello|hello|[-0.030862354, -0...|
## | test| test|[-0.13154022, 0.2...|
## +-----+-----+--------------------+

如果数据适合驱动程序/工作程序内存，则可以尝试通过广播进行收集和映射：

lookup_bd = sc.broadcast(lookup.rdd.collectAsMap())
rdd = sc.parallelize([['Hello'],['test']])
rdd.map(lambda ws: [lookup_bd.value.get(w) for w in ws])