apache-spark-sql

How do I convert an array (i.e. list) column to Vector

by

Personally I would go with Python UDF and wouldn’t bother with anything else: Vectors are not native SQL types so there will be performance overhead one way or another. In particular this process requires two steps where data is first converted from external type to row, and then from row to internal representation using generic … Read more

Spark functions vs UDF performance?

by

when would a udf be faster If you ask about Python UDF the answer is probably never*. Since SQL functions are relatively simple and are not designed for complex tasks it is pretty much impossible compensate the cost of repeated serialization, deserialization and data movement between Python interpreter and JVM. Does anyone know why this … Read more

Rename more than one column using withColumnRenamed

by

It is not possible to use a single withColumnRenamed call. You can use DataFrame.toDF method* data.toDF(‘x3’, ‘x4’) or new_names = [‘x3’, ‘x4’] data.toDF(*new_names) It is also possible to rename with simple select: from pyspark.sql.functions import col mapping = dict(zip([‘x1’, ‘x2’], [‘x3’, ‘x4’])) data.select([col(c).alias(mapping.get(c, c)) for c in data.columns]) Similarly in Scala you can: Rename all … Read more

How to convert column with string type to int form in pyspark data frame?

by

from pyspark.sql.types import IntegerType data_df = data_df.withColumn(“Plays”, data_df[“Plays”].cast(IntegerType())) data_df = data_df.withColumn(“drafts”, data_df[“drafts”].cast(IntegerType())) You can run loop for each column but this is the simplest way to convert string column into integer.

Median / quantiles within PySpark groupBy

by

I guess you don’t need it anymore. But will leave it here for future generations (i.e. me next week when I forget). from pyspark.sql import Window import pyspark.sql.functions as F grp_window = Window.partitionBy(‘grp’) magic_percentile = F.expr(‘percentile_approx(val, 0.5)’) df.withColumn(‘med_val’, magic_percentile.over(grp_window)) Or to address exactly your question, this also works: df.groupBy(‘grp’).agg(magic_percentile.alias(‘med_val’)) And as a bonus, you can … Read more

Querying Spark SQL DataFrame with complex types

by

It depends on a type of the column. Lets start with some dummy data: import org.apache.spark.sql.functions.{udf, lit} import scala.util.Try case class SubRecord(x: Int) case class ArrayElement(foo: String, bar: Int, vals: Array[Double]) case class Record( an_array: Array[Int], a_map: Map[String, String], a_struct: SubRecord, an_array_of_structs: Array[ArrayElement]) val df = sc.parallelize(Seq( Record(Array(1, 2, 3), Map(“foo” -> “bar”), SubRecord(1), Array( … Read more

Take n rows from a spark dataframe and pass to toPandas()

by

You can use the limit(n) function: l = [(‘Alice’, 1),(‘Jim’,2),(‘Sandra’,3)] df = sqlContext.createDataFrame(l, [‘name’, ‘age’]) df.limit(2).withColumn(‘age2’, df.age + 2).toPandas() Or: l = [(‘Alice’, 1),(‘Jim’,2),(‘Sandra’,3)] df = sqlContext.createDataFrame(l, [‘name’, ‘age’]) df.withColumn(‘age2’, df.age + 2).limit(2).toPandas()

How to perform union on two DataFrames with different amounts of columns in Spark?

by

In Scala you just have to append all missing columns as nulls. import org.apache.spark.sql.functions._ // let df1 and df2 the Dataframes to merge val df1 = sc.parallelize(List( (50, 2), (34, 4) )).toDF(“age”, “children”) val df2 = sc.parallelize(List( (26, true, 60000.00), (32, false, 35000.00) )).toDF(“age”, “education”, “income”) val cols1 = df1.columns.toSet val cols2 = df2.columns.toSet val … Read more