apache-spark-sql

Filtering a spark dataframe based on date

by

The following solutions are applicable since spark 1.5 : For lower than : // filter data where the date is lesser than 2015-03-14 data.filter(data(“date”).lt(lit(“2015-03-14”))) For greater than : // filter data where the date is greater than 2015-03-14 data.filter(data(“date”).gt(lit(“2015-03-14”))) For equality, you can use either equalTo or === : data.filter(data(“date”) === lit(“2015-03-14”)) If your DataFrame … Read more

PySpark create new column with mapping from a dict

by

Inefficient solution with UDF (version independent): from pyspark.sql.types import StringType from pyspark.sql.functions import udf def translate(mapping): def translate_(col): return mapping.get(col) return udf(translate_, StringType()) df = sc.parallelize([(‘DS’, ), (‘G’, ), (‘INVALID’, )]).toDF([‘key’]) mapping = { ‘A’: ‘S’, ‘B’: ‘S’, ‘C’: ‘S’, ‘DS’: ‘S’, ‘DNS’: ‘S’, ‘E’: ‘NS’, ‘F’: ‘NS’, ‘G’: ‘NS’, ‘H’: ‘NS’} df.withColumn(“value”, translate(mapping)(“key”)) with … Read more

How do I detect if a Spark DataFrame has a column

by

Just assume it exists and let it fail with Try. Plain and simple and supports an arbitrary nesting: import scala.util.Try import org.apache.spark.sql.DataFrame def hasColumn(df: DataFrame, path: String) = Try(df(path)).isSuccess val df = sqlContext.read.json(sc.parallelize( “””{“foo”: [{“bar”: {“foobar”: 3}}]}””” :: Nil)) hasColumn(df, “foobar”) // Boolean = false hasColumn(df, “foo”) // Boolean = true hasColumn(df, “foo.bar”) // Boolean … Read more

Find maximum row per group in Spark DataFrame

by

Using join (it will result in more than one row in group in case of ties): import pyspark.sql.functions as F from pyspark.sql.functions import count, col cnts = df.groupBy(“id_sa”, “id_sb”).agg(count(“*”).alias(“cnt”)).alias(“cnts”) maxs = cnts.groupBy(“id_sa”).agg(F.max(“cnt”).alias(“mx”)).alias(“maxs”) cnts.join(maxs, (col(“cnt”) == col(“mx”)) & (col(“cnts.id_sa”) == col(“maxs.id_sa”)) ).select(col(“cnts.id_sa”), col(“cnts.id_sb”)) Using window functions (will drop ties): from pyspark.sql.functions import row_number from pyspark.sql.window import … Read more

Spark SQL Row_number() PartitionBy Sort Desc

by

desc should be applied on a column not a window definition. You can use either a method on a column: from pyspark.sql.functions import col, row_number from pyspark.sql.window import Window F.row_number().over( Window.partitionBy(“driver”).orderBy(col(“unit_count”).desc()) ) or a standalone function: from pyspark.sql.functions import desc from pyspark.sql.window import Window F.row_number().over( Window.partitionBy(“driver”).orderBy(desc(“unit_count”)) )

Fetching distinct values on a column using Spark DataFrame

by

Well to obtain all different values in a Dataframe you can use distinct. As you can see in the documentation that method returns another DataFrame. After that you can create a UDF in order to transform each record. For example: val df = sc.parallelize(Array((1, 2), (3, 4), (1, 6))).toDF(“age”, “salary”) // I obtain all different … Read more

Including null values in an Apache Spark Join

by

Spark provides a special NULL safe equality operator: numbersDf .join(lettersDf, numbersDf(“numbers”) <=> lettersDf(“numbers”)) .drop(lettersDf(“numbers”)) +——-+——-+ |numbers|letters| +——-+——-+ | 123| abc| | 456| def| | null| zzz| | | hhh| +——-+——-+ Be careful not to use it with Spark 1.5 or earlier. Prior to Spark 1.6 it required a Cartesian product (SPARK-11111 – Fast null-safe join). … Read more

spark dataframe drop duplicates and keep first

by

To everyone saying that dropDuplicates keeps the first occurrence – this is not strictly correct. dropDuplicates keeps the ‘first occurrence’ of a sort operation – only if there is 1 partition. See below for some examples. However this is not practical for most Spark datasets. So I’m also including an example of ‘first occurrence’ drop … Read more