PySpark вставляет константу SparseVector в столбец Dataframe - PullRequest
Новая
       30

PySpark вставляет константу SparseVector в столбец Dataframe

0 голосов
/ 15 января 2019

Я хочу вставить в свой фрейм данных tfIdfFr столбец с именем "ref" с константой, тип которой pyspark.ml.linalg.SparseVector.

Когда я пытаюсь это 

ref = tfidfTest.select("features").collect()[0].features # the reference
tfIdfFr.withColumn("ref", ref).select("ref", "features").show()

Я получаю эту ошибку AssertionError: col should be Column

И когда я попробую это: 

from pyspark.sql.functions import lit
tfIdfFr.withColumn("ref", lit(ref)).select("ref", "features").show()

Я получаю эту ошибку AttributeError: 'SparseVector' object has no attribute '_get_object_id'

Знаете ли вы решение для вставки константы SparseVector в столбец Dataframe? *

1 Ответ

0 голосов
/ 15 января 2019

В этом случае вы просто пропустите сбор: 

ref = tfidfTest.select(col("features").alias("ref")).limit(1)
tfIdfFr.crossJoin(ref)

Как правило, вы можете использовать udf: 

from pyspark.ml.linalg import DenseVector, SparseVector, Vector, Vectors, \
 VectorUDT 
from pyspark.sql.functions import udf

def vector_lit(v): 
    assert isinstance(v, Vector) 
    return udf(lambda: v, VectorUDT())()

Использование: 

spark.range(1).select(
  vector_lit(Vectors.sparse(5, [1, 3], [-1, 1])
).alias("ref")).show()

+--------------------+
|                 ref|
+--------------------+
|(5,[1,3],[-1.0,1.0])|
+--------------------+

spark.range(1).select(vector_lit(Vectors.dense([1, 2, 3])).alias("ref")).show() 

+-------------+
|          ref|
+-------------+
|[1.0,2.0,3.0]|
+-------------+

Также можно использовать промежуточное представление: 

import json
from pyspark.sql.functions import from_json, lit
from pyspark.sql.types import StructType, StructField

def as_column(v):
    assert isinstance(v, Vector) 
    if isinstance(v, DenseVector):
        j = lit(json.dumps({"v": {
          "type": 1,
          "values": v.values.tolist()
        }}))
    else:
        j = lit(json.dumps({"v": {
          "type": 0,
          "size": v.size,
          "indices": v.indices.tolist(),
          "values": v.values.tolist()
        }}))
    return from_json(j, StructType([StructField("v", VectorUDT())]))["v"]

Использование: 

spark.range(1).select(
    as_column(Vectors.sparse(5, [1, 3], [-1, 1])
 ).alias("ref")).show()  

+--------------------+
|                 ref|
+--------------------+
|(5,[1,3],[-1.0,1.0])|
+--------------------+

spark.range(1).select(as_column(Vectors.dense([1, 2, 3])).alias("ref")).show()

+-------------+
|          ref|
+-------------+
|[1.0,2.0,3.0]|
+-------------+
...