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

【问题标题】：UDF to map words to term Index in SparkUDF 将单词映射到 Spark 中的术语索引
【发布时间】：2017-07-06 04:44:41
【问题描述】：

我正在尝试从 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)

编辑：

所以我想到了用mapper函数代替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'

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql user-defined-functions apache-spark-ml

【解决方案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() 方法。

【讨论】：

【解决方案2】：

如果可以，您想使用StringIndexer 和IndexToString。 StringIndexer 将从一列术语中为您创建术语索引，IndexToString 将在给定索引的情况下查找字符串值（您的 *term_to_words* 函数）。链接的 Spark 文档中有代码示例。

【讨论】：