scala - ¿Cómo pivotar Spark DataFrame?
apache-spark apache-spark-sql (9)
Hay muchos ejemplos de operaciones dinámicas en el conjunto de datos / marco de datos, pero no pude encontrar muchos usando SQL. Aquí hay un ejemplo que funcionó para mí.
# pass an optional list of string to avoid computation of columns
def pivot(df, group_by, key, aggFunction, levels=[]):
if not levels:
levels = [row[key] for row in df.filter(col(key).isNotNull()).groupBy(col(key)).agg(count(key)).select(key).collect()]
return df.filter(col(key).isin(*levels) == True).groupBy(group_by).agg(map_from_entries(collect_list(struct(key, expr(aggFunction)))).alias("group_map")).select([group_by] + ["group_map." + l for l in levels])
# Usage
pivot(df, "id", "key", "value")
pivot(df, "id", "key", "array(value)")
Estoy empezando a usar Spark DataFrames y necesito poder pivotar los datos para crear múltiples columnas de 1 columna con múltiples filas. Hay una funcionalidad integrada para eso en Scalding y creo en Pandas en Python, pero no puedo encontrar nada para el nuevo Spark Dataframe.
Supongo que puedo escribir una función personalizada de algún tipo que haga esto, pero ni siquiera estoy seguro de cómo comenzar, especialmente porque soy un novato con Spark. Si alguien sabe cómo hacer esto con la funcionalidad integrada o sugerencias sobre cómo escribir algo en Scala, es muy apreciado.
Hay una solución simple y elegante.
import java.util.Date
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.StructField
...
...
def main(args: Array[String]): Unit = {
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val dfdata1 = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema", "true")
.load("data.csv")
dfdata1.show()
val dfOutput = transpose(new HiveContext(sc), dfdata1, List("id"), "tag", "value")
dfOutput.show
}
He resuelto un problema similar usando marcos de datos con los siguientes pasos:
Cree columnas para todos sus países, con ''valor'' como valor:
import org.apache.spark.sql.functions._
val countries = List("US", "UK", "Can")
val countryValue = udf{(countryToCheck: String, countryInRow: String, value: Long) =>
if(countryToCheck == countryInRow) value else 0
}
val countryFuncs = countries.map{country => (dataFrame: DataFrame) => dataFrame.withColumn(country, countryValue(lit(country), df("tag"), df("value"))) }
val dfWithCountries = Function.chain(countryFuncs)(df).drop("tag").drop("value")
Su marco de datos ''dfWithCountries'' se verá así:
+--+--+---+---+
|id|US| UK|Can|
+--+--+---+---+
| 1|50| 0| 0|
| 1| 0|100| 0|
| 1| 0| 0|125|
| 2|75| 0| 0|
| 2| 0|150| 0|
| 2| 0| 0|175|
+--+--+---+---+
Ahora puede sumar todos los valores para el resultado deseado:
dfWithCountries.groupBy("id").sum(countries: _*).show
Resultado:
+--+-------+-------+--------+
|id|SUM(US)|SUM(UK)|SUM(Can)|
+--+-------+-------+--------+
| 1| 50| 100| 125|
| 2| 75| 150| 175|
+--+-------+-------+--------+
Sin embargo, no es una solución muy elegante. Tuve que crear una cadena de funciones para agregar en todas las columnas. Además, si tengo muchos países, expandiré mi conjunto de datos temporales a un conjunto muy amplio con muchos ceros.
Inicialmente adopté la solución de Al M. Más tarde tomó el mismo pensamiento y reescribió esta función como una función de transposición.
Este método transpone cualquier fila de df a columnas de cualquier formato de datos con el uso de la columna de clave y valor
para entrada csv
scala> spark.sql("select * from k_tags limit 10").show()
+---------------+-------------+------+
| imsi| name| value|
+---------------+-------------+------+
|246021000000000| age| 37|
|246021000000000| gender|Female|
|246021000000000| arpu| 22|
|246021000000000| DeviceType| Phone|
|246021000000000|DataAllowance| 6GB|
+---------------+-------------+------+
scala> spark.sql("select * from k_tags limit 10").groupBy($"imsi").pivot("name").agg(min($"value")).show()
+---------------+-------------+----------+---+----+------+
| imsi|DataAllowance|DeviceType|age|arpu|gender|
+---------------+-------------+----------+---+----+------+
|246021000000000| 6GB| Phone| 37| 22|Female|
|246021000000001| 1GB| Phone| 72| 10| Male|
+---------------+-------------+----------+---+----+------+
salida
id,tag,value
1,US,50a
1,UK,100
1,Can,125
2,US,75
2,UK,150
2,Can,175
método de transposición:
+--+---+---+---+
|id| UK| US|Can|
+--+---+---+---+
| 2|150| 75|175|
| 1|100|50a|125|
+--+---+---+---+
fragmento principal
def transpose(hc : HiveContext , df: DataFrame,compositeId: List[String], key: String, value: String) = {
val distinctCols = df.select(key).distinct.map { r => r(0) }.collect().toList
val rdd = df.map { row =>
(compositeId.collect { case id => row.getAs(id).asInstanceOf[Any] },
scala.collection.mutable.Map(row.getAs(key).asInstanceOf[Any] -> row.getAs(value).asInstanceOf[Any]))
}
val pairRdd = rdd.reduceByKey(_ ++ _)
val rowRdd = pairRdd.map(r => dynamicRow(r, distinctCols))
hc.createDataFrame(rowRdd, getSchema(df.schema, compositeId, (key, distinctCols)))
}
private def dynamicRow(r: (List[Any], scala.collection.mutable.Map[Any, Any]), colNames: List[Any]) = {
val cols = colNames.collect { case col => r._2.getOrElse(col.toString(), null) }
val array = r._1 ++ cols
Row(array: _*)
}
private def getSchema(srcSchema: StructType, idCols: List[String], distinctCols: (String, List[Any])): StructType = {
val idSchema = idCols.map { idCol => srcSchema.apply(idCol) }
val colSchema = srcSchema.apply(distinctCols._1)
val colsSchema = distinctCols._2.map { col => StructField(col.asInstanceOf[String], colSchema.dataType, colSchema.nullable) }
StructType(idSchema ++ colsSchema)
}
La función de pivote de chispa incorporada es ineficiente. La siguiente implementación funciona en spark 2.4+: la idea es agregar un mapa y extraer los valores como columnas. La única limitación es que no maneja la función de agregado en las columnas pivotantes, solo las columnas.
En una tabla de 8M, esas funciones se aplican en 3 segundos , frente a 40 minutos en la versión de chispa incorporada:
// pass an optional list of string to avoid computation of columns
def pivot(df: DataFrame, groupBy: Column, key: Column, aggFunct: String, _levels: List[String] = Nil): DataFrame = {
val levels =
if (_levels.isEmpty) df.filter(key.isNotNull).select(key).distinct().collect().map(row => row.getString(0)).toList
else _levels
df
.filter(key.isInCollection(levels))
.groupBy(groupBy)
.agg(map_from_entries(collect_list(struct(key, expr(aggFunct)))).alias("group_map"))
.select(groupBy.toString, levels.map(f => "group_map." + f): _*)
}
// Usage:
pivot(df, col("id"), col("key"), "value")
pivot(df, col("id"), col("key"), "array(value)")
// pass an optional list of string to avoid computation of columns def pivot(df: DataFrame, groupBy: Column, key: Column, aggFunct: String, _levels: List[String] = Nil): DataFrame = { val levels = if (_levels.isEmpty) df.filter(key.isNotNull).select(key).distinct().collect().map(row => row.getString(0)).toList else _levels df .filter(key.isInCollection(levels)) .groupBy(groupBy) .agg(map_from_entries(collect_list(struct(key, expr(aggFunct)))).alias("group_map")) .select(groupBy.toString, levels.map(f => "group_map." + f): _*) } // Usage: pivot(df, col("id"), col("key"), "value") pivot(df, col("id"), col("key"), "array(value)")
Lo superé escribiendo un bucle for para crear dinámicamente una consulta SQL. Digamos que tengo:
id tag value
1 US 50
1 UK 100
1 Can 125
2 US 75
2 UK 150
2 Can 175
y yo quiero:
id US UK Can
1 50 100 125
2 75 150 175
Puedo crear una lista con el valor que quiero pivotar y luego crear una cadena que contenga la consulta SQL que necesito.
val countries = List("US", "UK", "Can")
val numCountries = countries.length - 1
var query = "select *, "
for (i <- 0 to numCountries-1) {
query += """case when tag = """" + countries(i) + """" then value else 0 end as """ + countries(i) + ", "
}
query += """case when tag = """" + countries.last + """" then value else 0 end as """ + countries.last + " from myTable"
myDataFrame.registerTempTable("myTable")
val myDF1 = sqlContext.sql(query)
Puedo crear una consulta similar para luego hacer la agregación. No es una solución muy elegante, pero funciona y es flexible para cualquier lista de valores, que también se puede pasar como argumento cuando se llama a su código.
Se ha agregado un operador de pivote a la API de marco de datos de Spark, y forma parte de Spark 1.6.
Consulte https://github.com/apache/spark/pull/7841 para más detalles.
Spark ha estado proporcionando mejoras a Pivoting the Spark DataFrame. Se ha agregado una función pivote a la API Spark DataFrame a la versión Spark 1.6 y tiene un problema de rendimiento que se ha corregido en Spark 2.0
sin embargo, si está utilizando una versión inferior; tenga en cuenta que pivote es una operación muy costosa, por lo tanto, se recomienda proporcionar datos de columna (si se conoce) como argumento para funcionar como se muestra a continuación.
val countries = Seq("USA","China","Canada","Mexico") val pivotDF = df.groupBy("Product").pivot("Country", countries).sum("Amount") pivotDF.show()
Esto se ha explicado detalladamente en Pivoting and Unpivoting Spark DataFrame
Feliz aprendizaje !!
Como lo menciona
David Anderson,
Spark proporciona la función
pivot
desde la versión 1.6.
La sintaxis general tiene el siguiente aspecto:
df
.groupBy(grouping_columns)
.pivot(pivot_column, [values])
.agg(aggregate_expressions)
Ejemplos de uso con el formato
nycflights13
y
csv
:
Python :
from pyspark.sql.functions import avg
flights = (sqlContext
.read
.format("csv")
.options(inferSchema="true", header="true")
.load("flights.csv")
.na.drop())
flights.registerTempTable("flights")
sqlContext.cacheTable("flights")
gexprs = ("origin", "dest", "carrier")
aggexpr = avg("arr_delay")
flights.count()
## 336776
%timeit -n10 flights.groupBy(*gexprs ).pivot("hour").agg(aggexpr).count()
## 10 loops, best of 3: 1.03 s per loop
Scala :
val flights = sqlContext
.read
.format("csv")
.options(Map("inferSchema" -> "true", "header" -> "true"))
.load("flights.csv")
flights
.groupBy($"origin", $"dest", $"carrier")
.pivot("hour")
.agg(avg($"arr_delay"))
Java :
import static org.apache.spark.sql.functions.*;
import org.apache.spark.sql.*;
Dataset<Row> df = spark.read().format("csv")
.option("inferSchema", "true")
.option("header", "true")
.load("flights.csv");
df.groupBy(col("origin"), col("dest"), col("carrier"))
.pivot("hour")
.agg(avg(col("arr_delay")));
R / SparkR :
library(magrittr)
flights <- read.df("flights.csv", source="csv", header=TRUE, inferSchema=TRUE)
flights %>%
groupBy("origin", "dest", "carrier") %>%
pivot("hour") %>%
agg(avg(column("arr_delay")))
R / sparklyr
library(dplyr)
flights <- spark_read_csv(sc, "flights", "flights.csv")
avg.arr.delay <- function(gdf) {
expr <- invoke_static(
sc,
"org.apache.spark.sql.functions",
"avg",
"arr_delay"
)
gdf %>% invoke("agg", expr, list())
}
flights %>%
sdf_pivot(origin + dest + carrier ~ hour, fun.aggregate=avg.arr.delay)
SQL :
Tenga en cuenta que la palabra clave PIVOT en Spark SQL es compatible a partir de la versión 2.4.
CREATE TEMPORARY VIEW flights
USING csv
OPTIONS (header ''true'', path ''flights.csv'', inferSchema ''true'') ;
SELECT * FROM (
SELECT origin, dest, carrier, arr_delay, hour FROM flights
) PIVOT (
avg(arr_delay)
FOR hour IN (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12,
13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23)
);
Datos de ejemplo :
"year","month","day","dep_time","sched_dep_time","dep_delay","arr_time","sched_arr_time","arr_delay","carrier","flight","tailnum","origin","dest","air_time","distance","hour","minute","time_hour"
2013,1,1,517,515,2,830,819,11,"UA",1545,"N14228","EWR","IAH",227,1400,5,15,2013-01-01 05:00:00
2013,1,1,533,529,4,850,830,20,"UA",1714,"N24211","LGA","IAH",227,1416,5,29,2013-01-01 05:00:00
2013,1,1,542,540,2,923,850,33,"AA",1141,"N619AA","JFK","MIA",160,1089,5,40,2013-01-01 05:00:00
2013,1,1,544,545,-1,1004,1022,-18,"B6",725,"N804JB","JFK","BQN",183,1576,5,45,2013-01-01 05:00:00
2013,1,1,554,600,-6,812,837,-25,"DL",461,"N668DN","LGA","ATL",116,762,6,0,2013-01-01 06:00:00
2013,1,1,554,558,-4,740,728,12,"UA",1696,"N39463","EWR","ORD",150,719,5,58,2013-01-01 05:00:00
2013,1,1,555,600,-5,913,854,19,"B6",507,"N516JB","EWR","FLL",158,1065,6,0,2013-01-01 06:00:00
2013,1,1,557,600,-3,709,723,-14,"EV",5708,"N829AS","LGA","IAD",53,229,6,0,2013-01-01 06:00:00
2013,1,1,557,600,-3,838,846,-8,"B6",79,"N593JB","JFK","MCO",140,944,6,0,2013-01-01 06:00:00
2013,1,1,558,600,-2,753,745,8,"AA",301,"N3ALAA","LGA","ORD",138,733,6,0,2013-01-01 06:00:00
Consideraciones de rendimiento :
En general, pivotar es una operación costosa.
-
si puede, intente proporcionar una lista de
values, ya que esto evita un golpe adicional para calcular los únicos:vs = list(range(25)) %timeit -n10 flights.groupBy(*gexprs ).pivot("hour", vs).agg(aggexpr).count() ## 10 loops, best of 3: 392 ms per loop -
en algunos casos demostró ser beneficioso (probablemente ya no valga la pena el esfuerzo en 2.0 o posterior ) para
repartitiony / o agregar previamente los datos -
solo para remodelar, puede usar
first: la columna Pivot String en Pyspark Dataframe
Preguntas relacionadas :
- ¿Cómo derretir Spark DataFrame?
- Desvincular en spark-sql / pyspark
- Transponer columna a fila con Spark