Requirements for converting Spark dataframe to Pandas/R dataframe

by

toPandas (PySpark) / as.data.frame (SparkR)

Data has to be collected before local data frame is created. For example toPandas method looks as follows:

def toPandas(self):
    import pandas as pd
    return pd.DataFrame.from_records(self.collect(), columns=self.columns)

You need Python, optimally with all the dependencies, installed on each node.

SparkR counterpart (as.data.frame) is simply an alias for collect.

To summarize in both cases data is collected to the driver node and converted to the local data structure (pandas.DataFrame and base::data.frame in Python and R respectively).

Vectorized user defined functions

Since Spark 2.3.0 PySpark also provides a set of pandas_udf (SCALAR, GROUPED_MAP, GROUPED_AGG) which operate in parallel on chunks of data defined by

  • Partitions in case of SCALAR variant
  • Grouping expression in case of GROUPED_MAP and GROUPED_AGG.

Each chunk is represented by

  • One or more pandas.core.series.Series in case of SCALAR and GROUPED_AGG variants.
  • A single pandas.core.frame.DataFrame in case of GROUPED_MAP variant.

Similarly, since Spark 2.0.0, SparkR provides dapply and gapply functions operating on data.frames defined by partitions and grouping expressions respectively.

Aforementioned functions:

  • Don’t collect to the driver. Unless data contains only a single partition (i.e. with coalesce(1)) or grouping expression is trivial (i.e. groupBy(lit(1))) there is no single node bottleneck.
  • Load respective chunks in memory of the corresponding executor. As a result it is limited by the size of individual chunks / memory available on each executor.

More Related Contents:

  1. Convert a spark DataFrame to pandas DF
  2. How to pivot Spark DataFrame?
  3. How to split data into 3 sets (train, validation and test)?
  4. Spark SQL: apply aggregate functions to a list of columns
  5. Spark – load CSV file as DataFrame?
  6. How to define partitioning of DataFrame?
  7. DataFrame join optimization – Broadcast Hash Join
  8. How do I add a new column to a Spark DataFrame (using PySpark)?
  9. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  10. Dropping a nested column from Spark DataFrame
  11. Filter Spark DataFrame based on another DataFrame that specifies denylist criteria
  12. Updating a dataframe column in spark
  13. Dividing complex rows of dataframe to simple rows in Pyspark
  14. Apache Spark how to append new column from list/array to Spark dataframe
  15. Pandas, group by count and add count to original dataframe?
  16. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  17. Spark Dataframes UPSERT to Postgres Table
  18. Create new Dataframe with empty/null field values
  19. Why does Spark think this is a cross / Cartesian join
  20. Merge two data frames based on common column values in Pandas
  21. Pyspark filter dataframe by columns of another dataframe
  22. How do I add an persistent column of row ids to Spark DataFrame?
  23. Converting Pandas dataframe into Spark dataframe error
  24. Convert null values to empty array in Spark DataFrame
  25. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  26. Splitting multiple columns into rows in pandas dataframe
  27. Filtering DataFrame using the length of a column
  28. How to exclude multiple columns in Spark dataframe in Python
  29. Why is predicate pushdown not used in typed Dataset API (vs untyped DataFrame API)?
  30. Difference between === null and isNull in Spark DataDrame

Leave a Comment