我正在尝试获取从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'