python - multiple - sparksql pivot

Transponer columna a fila con Spark

Question

Estoy tratando de transponer algunas columnas de mi tabla a la fila. Estoy usando Python y Spark 1.5.0. Aquí está mi tabla inicial:

+-----+-----+-----+-------+
| A |col_1|col_2|col_...|
+-----+-------------------+
| 1 | 0.0| 0.6| ... |
| 2 | 0.6| 0.7| ... |
| 3 | 0.5| 0.9| ... |
| ...| ...| ...| ... |

Me gustaría tener algo como esto:

+-----+--------+-----------+
| A | col_id | col_value |
+-----+--------+-----------+
| 1 | col_1| 0.0|
| 1 | col_2| 0.6|
| ...| ...| ...|
| 2 | col_1| 0.6|
| 2 | col_2| 0.7|
| ...| ...| ...|
| 3 | col_1| 0.5|
| 3 | col_2| 0.9|
| ...| ...| ...|

¿Alguien sabe cómo puedo hacerlo? Gracias por tu ayuda.

Answer 1

Es relativamente simple de hacer con las funciones básicas de Spark SQL.

Pitón

from pyspark.sql.functions import array, col, explode, struct, lit
df = sc.parallelize([(1, 0.0, 0.6), (1, 0.6, 0.7)]).toDF(["A", "col_1", "col_2"])
def to_long(df, by):
# Filter dtypes and split into column names and type description
cols, dtypes = zip(*((c, t) for (c, t) in df.dtypes if c not in by))
# Spark SQL supports only homogeneous columns
assert len(set(dtypes)) == 1, "All columns have to be of the same type"
# Create and explode an array of (column_name, column_value) structs
kvs = explode(array([
struct(lit(c).alias("key"), col(c).alias("val")) for c in cols
])).alias("kvs")
return df.select(by + [kvs]).select(by + ["kvs.key", "kvs.val"])
to_long(df, ["A"])

Scala :

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{array, col, explode, lit, struct}
val df = Seq((1, 0.0, 0.6), (1, 0.6, 0.7)).toDF("A", "col_1", "col_2")
def toLong(df: DataFrame, by: Seq[String]): DataFrame = {
val (cols, types) = df.dtypes.filter{ case (c, _) => !by.contains(c)}.unzip
require(types.distinct.size == 1, s"${types.distinct.toString}.length != 1")
val kvs = explode(array(
cols.map(c => struct(lit(c).alias("key"), col(c).alias("val"))): _*
))
val byExprs = by.map(col(_))
df
.select(byExprs :+ kvs.alias("_kvs"): _*)
.select(byExprs ++ Seq($"_kvs.key", $"_kvs.val"): _*)
}
toLong(df, Seq("A"))

Answer 2

Las bibliotecas locales de álgebra lineal de Spark son actualmente muy débiles: y no incluyen operaciones básicas como las anteriores.

Existe una JIRA para arreglar esto para Spark 2.1, pero eso no lo ayudará hoy .

Algo a considerar: realizar una transposición probablemente requerirá mezclar completamente los datos.

Por ahora necesitará escribir el código RDD directamente. He escrito transpose en scala, pero no en python. Aquí está la versión scala :

 def transpose(mat: DMatrix) = {
val nCols = mat(0).length
val matT = mat
.flatten
.zipWithIndex
.groupBy {
_._2 % nCols
}
.toSeq.sortBy {
_._1
}
.map(_._2)
.map(_.map(_._1))
.toArray
matT
}

Entonces puede convertir eso a Python para su uso. No tengo ancho de banda para escribir / probar eso en este momento en particular: avíseme si no pudo hacer esa conversión.

Como mínimo, los siguientes se convierten fácilmente a python .

• zipWithIndex -> enumerate() (equivalente a python - crédito a @ zero323)
• map -> [someOperation(x) for x in ..]
• groupBy -> itertools.groupBy()

Aquí está la implementación de flatten que no tiene un equivalente en python:

 def flatten(L):
for item in L:
try:
for i in flatten(item):
yield i
except TypeError:
yield item

Por lo tanto, debería poder juntarlos para encontrar una solución.

Answer 3

Tomé la respuesta de Scala que escribió @javadba y creé una versión de Python para transponer todas las columnas en un DataFrame . Esto podría ser un poco diferente de lo que OP estaba preguntando ...

from itertools import chain
from pyspark.sql import DataFrame
def _sort_transpose_tuple(tup):
x, y = tup
return x, tuple(zip(*sorted(y, key=lambda v_k: v_k[1], reverse=False)))[0]
def transpose(X):
"""Transpose a PySpark DataFrame.
Parameters
----------
X : PySpark ``DataFrame``
The ``DataFrame`` that should be tranposed.
"""
# validate
if not isinstance(X, DataFrame):
raise TypeError(''X should be a DataFrame, not a %s''
% type(X))
cols = X.columns
n_features = len(cols)
# Sorry for this unreadability...
return X.rdd.flatMap( # make into an RDD
lambda xs: chain(xs)).zipWithIndex().groupBy( # zip index
lambda val_idx: val_idx[1] % n_features).sortBy( # group by index % n_features as key
lambda grp_res: grp_res[0]).map( # sort by index % n_features key
lambda grp_res: _sort_transpose_tuple(grp_res)).map( # maintain order
lambda key_col: key_col[1]).toDF() # return to DF

Por ejemplo:

>>> X = sc.parallelize([(1,2,3), (4,5,6), (7,8,9)]).toDF()
>>> X.show()
+---+---+---+
| _1| _2| _3|
+---+---+---+
| 1| 2| 3|
| 4| 5| 6|
| 7| 8| 9|
+---+---+---+
>>> transpose(X).show()
+---+---+---+
| _1| _2| _3|
+---+---+---+
| 1| 4| 7|
| 2| 5| 8|
| 3| 6| 9|
+---+---+---+

Answer 4

Una forma de resolver con pyspark sql usando las funciones create_map y explode .

from pyspark.sql import functions as func
#Use `create_map` to create the map of columns with constant
df = df.withColumn(''mapCol'', /
func.create_map(func.lit(''col_1''),df.col_1,
func.lit(''col_2''),df.col_2,
func.lit(''col_3''),df.col_3
)
)
#Use explode function to explode the map
res = df.select(''*'',func.explode(df.mapCol).alias(''col_id'',''col_value''))
res.show()

Answer 5

Una forma muy práctica de implementar:

from pyspark.sql import Row
def rowExpander(row):
rowDict = row.asDict()
valA = rowDict.pop(''A'')
for k in rowDict:
yield Row(**{''A'': valA , ''colID'' : k, ''colValue'' : row[k]})
newDf = sqlContext.createDataFrame(df.rdd.flatMap(rowExpander)

Answer 6

Utiliza el mapa plano. Algo como a continuación debería funcionar

from pyspark.sql import Row
def rowExpander(row):
rowDict = row.asDict()
valA = rowDict.pop(''A'')
for k in rowDict:
yield Row(**{''A'': valA , ''colID'': k, ''colValue'': row[k]})
newDf = sqlContext.createDataFrame(df.rdd.flatMap(rowExpander))