Cómo pivotar Spark DataFrame?

Estoy empezando a utilizar Spark Dataframes y necesito poder pivotar los datos para crear varias columnas de una columna con varias filas. Hay una funcionalidad incorporada para eso en Scalding y creo en Pandas en python, pero no puedo encontrar nada para el nuevo Spark Dataframe.

Supongo que puedo escribir funciones personalizadas de algún tipo que harán 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 funcionalidad incorporada o sugerencias sobre cómo escribir algo en Scala, es muy apreciado.

Como lo menciona David Anderson, Spark proporciona la función de pivot desde la versión 1.6. La syntax general se ve de la siguiente manera:

 df .groupBy(grouping_columns) .pivot(pivot_column, [values]) .agg(aggregate_expressions) 

Ejemplos de uso que utilizan 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 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) 

Ejemplo de datos :

 "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 intentar proporcionar una lista de values :

     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, resultó ser beneficioso para repartition y / o agregar previamente los datos

  • solo para la remodelación, puede usar first : columna Pivot String en Pyspark Dataframe

Preguntas relacionadas

  • Cómo derretir Spark DataFrame?
  • unpivot en spark-sql / pyspark
  • Transponer la columna a la fila con Spark

Superé esto 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 puede transmitirse como argumento cuando se invoca su código.

Se ha agregado un operador de pivote a la API de dataframe de Spark, y es parte de Spark 1.6.

Visita https://github.com/apache/spark/pull/7841 para más detalles.

He resuelto un problema similar usando dataframes con los siguientes pasos:

Crea columnas para todos tus 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 dataframe ‘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 sumr 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| +--+-------+-------+--------+ 

Aunque 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 las columnas de cualquier formato de datos con el uso de clave y columna de valor

para entrada csv

 id,tag,value 1,US,50a 1,UK,100 1,Can,125 2,US,75 2,UK,150 2,Can,175 

salida

 +--+---+---+---+ |id| UK| US|Can| +--+---+---+---+ | 2|150| 75|175| | 1|100|50a|125| +--+---+---+---+ 

método de transposición:

 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) } 

fragmento principal

 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 } 

Hay una solución simple y elegante.

 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| +---------------+-------------+----------+---+----+------+