Spark中将单词映射到术语索引的UDF

Question

我正在尝试获取从LDA模型获得的术语ID的相应主题词。

以下是在Spark中从LDA获取的主题及其单词分布的数据框：

topics_desc=ldaModel.describeTopics(20)
topics_desc.show(1)

+-----+--------------------+--------------------+
|topic|         termIndices|         termWeights|
+-----+--------------------+--------------------+
|    0|[0, 39, 68, 43, 5...|[0.06362107696025...|
+-----+--------------------+--------------------+
only showing top 1 row

现在我们有了termIndices而不是实际的单词，我想在这个数据框中添加另一列，对应于termIndices的单词。

现在，由于我在Spark中运行了CountVectorizer，所以我使用该模型并获取以下单词数组列表。

# Creating Term Frequency Vector for each word
cv=CountVectorizer(inputCol="words", outputCol="tf_features", minDF=2.0)
cvModel=cv.fit(swremoved_df)

cvModel.vocabulary 返回单词列表。

下面是我编写的一个UDF，用于获取映射：

from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType

def term_to_words(termindices):
    """ To get the corresponding words from term indices

    """


    return np.array(cvModel.vocabulary)[termindices]

term_to_words_conv=udf(term_to_words)


topics=topics_desc.withColumn("topics_words",term_to_words_conv("termIndices"))

我将列表转换为np数组的原因是，在numpy数组中，我可以通过传递索引列表来进行索引，而在列表中无法这样做。

但是，我遇到了这个错误。我不确定为什么会出现这种情况，因为我几乎什么都没有做。

Py4JError: An error occurred while calling o443.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
    at py4j.Gateway.invoke(Gateway.java:272)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:214)
    at java.lang.Thread.run(Thread.java:745)

编辑：

因此，我考虑使用映射器函数而不是UDF。

def term_to_words(x):
    """ Mapper function to get the corresponding words for the term index

    """

    row=x.asDict()
    word_list=np.array(cvModel.vocabulary)

    return (row['topic'],row['termIndices'],row['termWeights'],word_list[row[termindices]])


topics_rdd=topics_desc.rdd.map(term_to_words)

/Users/spark2/python/pyspark/context.pyc in runJob(self, rdd, partitionFunc, partitions, allowLocal)
    931         # SparkContext#runJob.
    932         mappedRDD = rdd.mapPartitions(partitionFunc)
--> 933         port = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions)
    934         return list(_load_from_socket(port, mappedRDD._jrdd_deserializer))
    935 

AttributeError: 'NoneType' object has no attribute 'sc'

- Baktaawar

Answer 1

这里有两个不同的问题：

• CountVectorizer是Java对象的包装器。它不能被序列化并通过闭包传递。出于同样的原因，您不能在map闭包中使用它。
• 您不能从UDF返回NumPy类型。

例如，您可以：

from pyspark.sql.types import ArrayType, StringType

def indices_to_terms(vocabulary):
    def indices_to_terms(xs):
        return [vocabulary[int(x)] for x in xs]
    return udf(indices_to_terms, ArrayType(StringType()))

使用方法：

topics_desc.withColumn(
    "topics_words", indices_to_terms(cvModel.vocabulary)("termIndices"))

如果你想使用NumPy数组，你需要在从UDF返回之前使用tolist()方法。

- zero323

Answer 2

如果可以的话，您应该使用StringIndexer和IndexToString。 StringIndexer将从术语列为您创建术语索引，而IndexToString将根据索引查找字符串值（即您的*term_to_words*函数）。链接的Spark文档中有代码示例。

- tmarthal