Ответ 1

Встроенной функции нет (если вы работаете с включенной поддержкой SQL и Hive, вы можете использовать stack функцию, но она не представлена в Spark и не имеет встроенной реализации), но ее тривиально накатить самостоятельно. Требуемый импорт:

from pyspark.sql.functions import array, col, explode, lit, struct
from pyspark.sql import DataFrame
from typing import Iterable 

Пример реализации:

def melt(
        df: DataFrame, 
        id_vars: Iterable[str], value_vars: Iterable[str], 
        var_name: str="variable", value_name: str="value") -> DataFrame:
    """Convert :class:'DataFrame' from wide to long format."""

    # Create array<struct<variable: str, value: ...>>
    _vars_and_vals = array(*(
        struct(lit(c).alias(var_name), col(c).alias(value_name)) 
        for c in value_vars))

    # Add to the DataFrame and explode
    _tmp = df.withColumn("_vars_and_vals", explode(_vars_and_vals))

    cols = id_vars + [
            col("_vars_and_vals")[x].alias(x) for x in [var_name, value_name]]
    return _tmp.select(*cols)

И некоторые тесты (основанные на doctests Pandas):

import pandas as pd

pdf = pd.DataFrame({'A': {0: 'a', 1: 'b', 2: 'c'},
                   'B': {0: 1, 1: 3, 2: 5},
                   'C': {0: 2, 1: 4, 2: 6}})

pd.melt(pdf, id_vars=['A'], value_vars=['B', 'C'])

   A variable  value
0  a        B      1
1  b        B      3
2  c        B      5
3  a        C      2
4  b        C      4
5  c        C      6

sdf = spark.createDataFrame(pdf)
melt(sdf, id_vars=['A'], value_vars=['B', 'C']).show()

+---+--------+-----+
|  A|variable|value|
+---+--------+-----+
|  a|       B|    1|
|  a|       C|    2|
|  b|       B|    3|
|  b|       C|    4|
|  c|       B|    5|
|  c|       C|    6|
+---+--------+-----+

Примечание. Для использования с устаревшими версиями Python удалите аннотации типов.

Связанные с:

• r sparkR - эквивалент функции расплава
• Соберитесь в sparklyr

Ответ 2

В этом вопросе пришел поиск в поиске melt в Spark для Scala.

Проводка моего порта Scala на случай, если кто-то тоже наткнется на это.

  import org.apache.spark.sql.functions._
import org.apache.spark.sql. {DataFrame}
/** Расширяет класс [[org.apache.spark.sql.DataFrame]]
 *
 * @param для фрейма данных для плавления
 */
неявный класс DataFrameFunctions (df: DataFrame) {
   /** Преобразовать [[org.apache.spark.sql.DataFrame]] из формата ширины в длинный.    *    * расплав является (видом) обратным стержнем    * расплав в настоящее время (02/2017) не реализован в искры    *    * @see reshape packe в R (https://cran.r-project.org/web/packages/reshape/index.html)    * @see это scala-адаптация http://stackoverflow.com/questions/41670103/pandas-melt-function-in-apache-spark    *    * перегрузка метода @todo для простого вызова    *    * @param id_vars столбцы для сохранения    * @param value_vars столбцы для таяния    * @param var_name имя столбца, содержащего имена расплавленных столбцов    * @param value_name имя столбца, содержащего значения расплавленных столбцов    *    */
   def расплава (           id_vars: Seq [String], value_vars: Seq [String],           var_name: String = "variable", value_name: String = "value" ): DataFrame = {
       // Создаем массив < struct < variable: str, value:... > >       val_vars_and_vals = array ((for (c < - value_vars) yield {struct (lit (c).alias(var_name), col (c).alias(value_name))}: _ *)
       // Добавить в DataFrame и взорваться       val _tmp = df.withColumn( "_ vars_and_vals", explode (_vars_and_vals))
       val cols = id_vars.map(col _) ++ {for (x < - List (var_name, value_name)) yield {col ( "_ vars_and_vals" ) (x).alias(x)}}
       return _tmp.select(cols: _ *)
   }
}
Код>

Поскольку я не настолько продвинутый, рассматривая Scala, я уверен, что есть возможности для улучшения.

Любые комментарии приветствуются.

Ответ 3

UPD

Наконец я нашел наиболее эффективную реализацию для меня. Он использует все ресурсы для кластера в моей конфигурации пряжи.

from pyspark.sql.functions import explode
def melt(df):
    sp = df.columns[1:]
    return (df
            .rdd
            .map(lambda x: [str(x[0]), [(str(i[0]), 
                                         float(i[1] if i[1] else 0)) for i in zip(sp, x[1:])]], 
                 preservesPartitioning = True)
            .toDF()
            .withColumn('_2', explode('_2'))
            .rdd.map(lambda x: [str(x[0]), 
                                str(x[1][0]), 
                                float(x[1][1] if x[1][1] else 0)], 
                     preservesPartitioning = True)
            .toDF()
            )

Для очень широкого массива данных у меня снижается производительность при генерации _vars_and_vals из ответа пользователя 6910411.

Было полезно реализовать плавление через selectExpr

columns=['a', 'b', 'c', 'd', 'e', 'f']
pd_df = pd.DataFrame([[1,2,3,4,5,6], [4,5,6,7,9,8], [7,8,9,1,2,4], [8,3,9,8,7,4]], columns=columns)
df = spark.createDataFrame(pd_df)
+---+---+---+---+---+---+
|  a|  b|  c|  d|  e|  f|
+---+---+---+---+---+---+
|  1|  2|  3|  4|  5|  6|
|  4|  5|  6|  7|  9|  8|
|  7|  8|  9|  1|  2|  4|
|  8|  3|  9|  8|  7|  4|
+---+---+---+---+---+---+

cols = df.columns[1:]
df.selectExpr('a', "stack({}, {})".format(len(cols), ', '.join(("'{}', {}".format(i, i) for i in cols))))
+---+----+----+
|  a|col0|col1|
+---+----+----+
|  1|   b|   2|
|  1|   c|   3|
|  1|   d|   4|
|  1|   e|   5|
|  1|   f|   6|
|  4|   b|   5|
|  4|   c|   6|
|  4|   d|   7|
|  4|   e|   9|
|  4|   f|   8|
|  7|   b|   8|
|  7|   c|   9|
...

Ответ 4

Проголосовал за ответ пользователя 6910411. Он работает, как и ожидалось, однако не может хорошо обрабатывать значения None. таким образом я реорганизовал его функцию расплава следующим образом:

from pyspark.sql.functions import array, col, explode, lit
from pyspark.sql.functions import create_map
from pyspark.sql import DataFrame
from typing import Iterable 
from itertools import chain

def melt(
        df: DataFrame, 
        id_vars: Iterable[str], value_vars: Iterable[str], 
        var_name: str="variable", value_name: str="value") -> DataFrame:
    """Convert :class:'DataFrame' from wide to long format."""

    # Create map<key: value>
    _vars_and_vals = create_map(
        list(chain.from_iterable([
            [lit(c), col(c)] for c in value_vars]
        ))
    )

    _tmp = df.select(*id_vars, explode(_vars_and_vals)) \
        .withColumnRenamed('key', var_name) \
        .withColumnRenamed('value', value_name)

    return _tmp

Тест проводится со следующим фреймом данных:

import pandas as pd

pdf = pd.DataFrame({'A': {0: 'a', 1: 'b', 2: 'c'},
                   'B': {0: 1, 1: 3, 2: 5},
                   'C': {0: 2, 1: 4, 2: 6},
                   'D': {1: 7, 2: 9}})

pd.melt(pdf, id_vars=['A'], value_vars=['B', 'C', 'D'])

A   variable    value
0   a   B   1.0
1   b   B   3.0
2   c   B   5.0
3   a   C   2.0
4   b   C   4.0
5   c   C   6.0
6   a   D   NaN
7   b   D   7.0
8   c   D   9.0

sdf = spark.createDataFrame(pdf)
melt(sdf, id_vars=['A'], value_vars=['B', 'C', 'D']).show()
+---+--------+-----+
|  A|variable|value|
+---+--------+-----+
|  a|       B|  1.0|
|  a|       C|  2.0|
|  a|       D|  NaN|
|  b|       B|  3.0|
|  b|       C|  4.0|
|  b|       D|  7.0|
|  c|       B|  5.0|
|  c|       C|  6.0|
|  c|       D|  9.0|
+---+--------+-----+