wake-up-neo.com

So teilen Sie Vector in Spalten - mit PySpark

Kontext: Ich habe ein DataFrame mit 2 Spalten: Wort und Vektor. Wobei der Spaltentyp von "Vektor" VectorUDT ist.

Ein Beispiel:

Word    |  vector
assert  | [435,323,324,212...]

Und ich möchte Folgendes bekommen:

Word   |  v1 | v2  | v3 | v4 | v5 | v6 ......
assert | 435 | 5435| 698| 356|....

Frage:

Wie kann ich mit PySpark eine Spalte mit Vektoren in mehrere Spalten für jede Dimension aufteilen?

Danke im Voraus

pythonapache-sparkpysparkapache-spark-sqlapache-spark-ml
32
sedioben

Ein möglicher Ansatz ist die Konvertierung von und nach RDD:

from pyspark.ml.linalg import Vectors

df = sc.parallelize([
    ("assert", Vectors.dense([1, 2, 3])),
    ("require", Vectors.sparse(3, {1: 2}))
]).toDF(["Word", "vector"])

def extract(row):
    return (row.Word, ) + Tuple(row.vector.toArray().tolist())

df.rdd.map(extract).toDF(["Word"])  # Vector values will be named _2, _3, ...

## +-------+---+---+---+
## |   Word| _2| _3| _4|
## +-------+---+---+---+
## | assert|1.0|2.0|3.0|
## |require|0.0|2.0|0.0|
## +-------+---+---+---+

Eine alternative Lösung wäre, eine UDF zu erstellen:

from pyspark.sql.functions import udf, col
from pyspark.sql.types import ArrayType, DoubleType

def to_array(col):
    def to_array_(v):
        return v.toArray().tolist()
    return udf(to_array_, ArrayType(DoubleType()))(col)

(df
    .withColumn("xs", to_array(col("vector")))
    .select(["Word"] + [col("xs")[i] for i in range(3)]))

## +-------+-----+-----+-----+
## |   Word|xs[0]|xs[1]|xs[2]|
## +-------+-----+-----+-----+
## | assert|  1.0|  2.0|  3.0|
## |require|  0.0|  2.0|  0.0|
## +-------+-----+-----+-----+

Für Scala Äquivalent siehe Spark Scala: Konvertieren von DataFrame [Vektor] in DataFrame [f1: Double, ..., fn: Double)] .

51
zero323
def splitVecotr(df, new_features=['f1','f2']):
schema = df.schema
cols = df.columns

for col in new_features: # new_features should be the same length as vector column length
    schema = schema.add(col,DoubleType(),True)

return spark.createDataFrame(df.rdd.map(lambda row: [row[i] for i in cols]+row.features.tolist()), schema)

Die Funktion wandelt die Merkmalsvektorspalte in separate Spalten um

0
Shuai Liu