Блог

Главная - Вопросы по программированию - Как получить верхний X слов из разреженного вектора в массив строк с помощью PySpark

Как получить верхний X слов из разреженного вектора в массив строк с помощью PySpark

#python #apache-spark #pyspark

#python #apache-spark #pyspark

Вопрос:

В настоящее время я кластеризую некоторые текстовые документы. Я использую K-means и обрабатываю свои данные с помощью TF-IDF благодаря методам PySpark. И теперь я хочу получить верхние 10 слов для каждого кластера :

Когда я делаю :

 getTopwords_udf = udf(lambda vector: [ countVectorizerModel.vocabulary[indice] for indice in  vector.toArray().tolist().argsort()[-10:][::-1]], ArrayType(StringType()))

predictions.groupBy("prediction").agg(Summarizer.mean(col("features")).alias("means")) 
    .withColumn("topWord", getTopwords_udf(col('means'))) 
    .select("prediction", "topWord") 
    .show(2, truncate=100)

Я получаю эту ошибку :

 Could not serialize object: Py4JError: An error occurred while calling o225.__getstate__. Trace:
py4j.Py4JException: Method __getstate__([]) 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:274)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)


Traceback (most recent call last):
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 189, in wrapper
    return self(*args)
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 167, in __call__
    judf = self._judf
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 151, in _judf
    self._judf_placeholder = self._create_judf()
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 160, in _create_judf
    wrapped_func = _wrap_function(sc, self.func, self.returnType)
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/sql/udf.py", line 35, in _wrap_function
    pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2420, in _prepare_for_python_RDD
    pickled_command = ser.dumps(command)
  File "/opt/bigpipe/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 597, in dumps
    raise pickle.PicklingError(msg)
_pickle.PicklingError: Could not serialize object: Py4JError: An error occurred while calling o225.__getstate__. Trace:
py4j.Py4JException: Method __getstate__([]) 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:274)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)

Я думал, что это из-за типа (от DoubleType до float для numpy), поэтому я тоже попробовал это, чтобы посмотреть, что происходит

 vector_udf = udf(lambda vector: vector.toArray().tolist(), ArrayType(FloatType()))
vector2_udf = udf(lambda vector: vector.sort()[:10], ArrayType(FloatType()))

predictions.groupBy("prediction").agg(Summarizer.mean(col("features")).alias("means")) 
    .withColumn("topWord", vector_udf(col('means'))) 
    .withColumn("topWord2", vector2_udf(col('topWord'))) 
    .select("prediction", "topWord", "topWord2") 
    .show(2, truncate=100)

Но я получаю эту ошибку TypeError: 'NoneType' object is not subscriptable

Комментарии:

1. одна потенциальная проблема может заключаться в том, vector_udf2 где у вас есть vector.sort()[:10] , поскольку vector.sort не возвращает значение.

2. Да, действительно, метод сортировки python ничего не возвращает: он напрямую изменяет список.

Ответ №1:

Я выяснил, как получить верхний X слов из разреженного вектора в массив строк с помощью PySpark. Вот мое решение для тех, кто может быть заинтересован…

 def getTopWordContainer(v):
    def getTopWord(vector):
        vectorConverted = vector.toArray().tolist()
        listSortedDesc= [i[0] for i in sorted(enumerate(vectorConverted), key=lambda x:x[1])][-10:][::-1]
        return [v[j] for j in listSortedDesc]
    return getTopWord

getTopWordInit = getTopWordContainer(countVectorizerModel.vocabulary)
getTopWord_udf = udf(getTopWordInit, ArrayType(StringType()))

top = predictions.groupBy("prediction").agg(Summarizer.mean(col("features")).alias("means")) 
    .withColumn("topWord", getTopWord_udf(col('means'))) 
    .select("prediction", "topWord")

Я новичок в spark, поэтому, если вы знаете, как его улучшить, дайте мне знать 🙂