How to split a dataframe into dataframes with same column values?

by

You can collect unique state values and simply map over resulting array:

val states = df.select("State").distinct.collect.flatMap(_.toSeq)
val byStateArray = states.map(state => df.where($"State" <=> state))

or to map:

val byStateMap = states
    .map(state => (state -> df.where($"State" <=> state)))
    .toMap

The same thing in Python:

from itertools import chain
from pyspark.sql.functions import col

states = chain(*df.select("state").distinct().collect())

# PySpark 2.3 and later
# In 2.2 and before col("state") == state) 
# should give the same outcome, ignoring NULLs 
# if NULLs are important 
# (lit(state).isNull() & col("state").isNull()) | (col("state") == state)
df_by_state = {state: 
  df.where(col("state").eqNullSafe(state)) for state in states}

The obvious problem here is that it requires a full data scan for each level, so it is an expensive operation. If you’re looking for a way to just split the output see also How do I split an RDD into two or more RDDs?

In particular you can write Dataset partitioned by the column of interest:

val path: String = ???
df.write.partitionBy("State").parquet(path)

and read back if needed:

// Depend on partition prunning
for { state <- states } yield spark.read.parquet(path).where($"State" === state)

// or explicitly read the partition
for { state <- states } yield spark.read.parquet(s"$path/State=$state")

Depending on the size of the data, number of levels of the splitting, storag and persistence level of the input it might faster or slower than multiple filters.

More Related Contents:

  1. How to query JSON data column using Spark DataFrames?
  2. How to define partitioning of DataFrame?
  3. How do I detect if a Spark DataFrame has a column
  4. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  5. Provide schema while reading csv file as a dataframe
  6. Dropping a nested column from Spark DataFrame
  7. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  8. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  9. How to zip two (or more) DataFrame in Spark
  10. Joining Spark dataframes on the key
  11. Apache Spark how to append new column from list/array to Spark dataframe
  12. Create new Dataframe with empty/null field values
  13. Provide schema while reading csv file as a dataframe in Scala Spark
  14. Renaming column names of a DataFrame in Spark Scala
  15. Derive multiple columns from a single column in a Spark DataFrame
  16. Replace missing values with mean – Spark Dataframe
  17. What is going wrong with `unionAll` of Spark `DataFrame`?
  18. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  19. How to get keys and values from MapType column in SparkSQL DataFrame
  20. Spark DataFrames when udf functions do not accept large enough input variables
  21. Filter spark DataFrame on string contains
  22. How to avoid duplicate columns after join?
  23. Flattening Rows in Spark
  24. How to create an empty DataFrame with a specified schema?
  25. Better way to convert a string field into timestamp in Spark
  26. how to filter out a null value from spark dataframe
  27. What are the various join types in Spark?
  28. How to calculate the size of dataframe in bytes in Spark?
  29. Spark-Monotonically increasing id not working as expected in dataframe?
  30. Difference between === null and isNull in Spark DataDrame

Leave a Comment