Создание PySpark Dataframe из вектора элементов с меткой

Question

У меня есть фрейм данных, который я создал с помощью объекта Pipeline, который выглядит следующим образом:

df.show()

+--------------------+-----+
|            features|label|
+--------------------+-----+
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
|[-0.0775219322931...|    0|
+--------------------+-----+

Я успешно извлек векторы объектов, как это:

df_table = df.rdd.map(lambda x: [float(y) for y in x['features']]).toDF(cols)

Проблема с выше, что он не сохраняет столбец метки. В качестве обходного пути я успешно использовал Join, чтобы вернуть столбец метки обратно, но я обнаружил, что он слишком запутанный.

Как бы я использовал однострочник, такой как приведенный выше, для извлечения вектора объектов и создания Вывести из него DF и одновременно добавить к нему и этот столбец меток?

Answer 1

У вас есть хорошие варианты здесь , особенно если у вас есть версия Spark> = 3.0.0

Если у вас нет такой последней версии, ваша проблема связана с тот факт, что вы потеряли свой ключ на вашей карте. Вы можете сделать:

df_table = df.rdd.map(lambda l: tuple([l['label']] + [float(y) for y in l['features']])).toDF()

В итоге вы получите широкоформатный фрейм данных. Если вам нужен длинный форматированный вектор, у вас есть больше опций.

Если вы хотите длинный форматированный данные

Сначала с rdd:

df.rdd.flatMapValues(lambda l: l).toDF(['label','feature'])

Или, еще лучше , напрямую используя DataFrame API: (непроверенное решение)

import pyspark.sql.functions as psf
df.select('label', psf.explode(psf.col('label')))