rdd.toDF() изменения всплывают None

Question

(Использование Apache Спарк 1.6.0) Привет всем,

У меня есть SparseVector (которая в основном определяется двумя Numpy массивами, а именно values и indices. Я хочу получить высокие значения и их индексы, чтобы сделать это я использую:

r = df.map(lambda row: Row(**dict(row.asDict(), top=f(vec))))) 

где функция возвращает е [ [sorted_indices], [sorted_values] ] следующим образом:

def f(v): 
    m, i = zip(*sorted(zip(v.values, v.indices), reverse=True)) 
    m = [ float(j) for j in m] 
    i = [ int(j) for j in i] 
    return [i, m] 

В этот момент r является pyspark.rdd.PipelinedRDD, и я могу проверить, что мои значения соответствуют используемым, например.

r.first().top[1] 

Проблема возникает, когда я пытаюсь получить DataFrame с помощью:

df2 = r.toDF() 

Тогда мои значения только None, т.е.

df2.first().top[1] # i.e. the highest values of the first Vector 

показывает None.

Итак, похоже, что функция toDF() уничтожает мои данные. Это было бы довольно странно, если Spark не смог обработать встроенный тип float.

Любые идеи? thx

Answer 1

Это не работает, потому что типы не совпадают. Если вы посмотрите на типы, вы увидите, что столбец top представлен как array<array<bigint>>, тогда как значения должны быть array<float>. Вы должны использовать объект, который может быть преобразован в столбец structstruct<array<bigint>, array<float>>. Очевидный выбор либо tuple или :

from pyspark.sql import Row 

def f(v): 
    m, i = zip(*sorted(zip(v.values, v.indices), reverse=True)) 
    m = [ float(j) for j in m] 
    i = [ int(j) for j in i] 
    return Row(indices=i, values=m) 

Кроме того, если вектор уже в DataFrame лучше использовать UDF здесь:

from pyspark.sql.functions import udf, col 
from pyspark.sql.types import * 

schema = StructType([ 
    StructField("indices", ArrayType(IntegerType())), 
    StructField("values", ArrayType(DoubleType())) 
]) 

df.withColumn("top", udf(f, schema)(col("vec_column")))