How to force DataFrame evaluation in Spark

by

It’s a bit late, but here’s the fundamental reason: count does not act the same on RDD and DataFrame.

In DataFrames there’s an optimization, as in some cases you do not require to load data to actually know the number of elements it has (especially in the case of yours where there’s no data shuffling involved). Hence, the DataFrame materialized when count is called will not load any data and will not pass into your exception throwing. You can easily do the experiment by defining your own DefaultSource and Relation and see that calling count on a DataFrame will always end up in the method buildScan with no requiredColumns no matter how many columns you did select (cf. org.apache.spark.sql.sources.interfaces to understand more). It’s actually a very efficient optimization 😉

In RDDs though, there’s no such optimizations (that’s why one should always try to use DataFrames when possible). Hence the count on RDD executes all the lineage and returns the sum of all sizes of the iterators composing any partitions.

Calling dataframe.count goes into the first explanation, but calling dataframe.rdd.count goes into the second as you did build an RDD out of your DataFrame. Note that calling dataframe.cache().count forces the dataframe to be materialized as you required Spark to cache the results (hence it needs to load all the data and transform it). But it does have the side-effect of caching your data…

More Related Contents:

  1. Encoder error while trying to map dataframe row to updated row
  2. How to define schema for custom type in Spark SQL?
  3. Specifying the filename when saving a DataFrame as a CSV [duplicate]
  4. NullPointerException in Scala Spark, appears to be caused be collection type?
  5. how to make saveAsTextFile NOT split output into multiple file?
  6. What does setMaster `local[*]` mean in spark?
  7. Spark unionAll multiple dataframes
  8. How can I pass extra parameters to UDFs in Spark SQL?
  9. How to split a dataframe into dataframes with same column values?
  10. Spark performance for Scala vs Python
  11. How to use regex to include/exclude some input files in sc.textFile?
  12. How to define a custom aggregation function to sum a column of Vectors?
  13. Spark DataFrame: does groupBy after orderBy maintain that order?
  14. Spark UDAF with ArrayType as bufferSchema performance issues
  15. Apache Spark how to append new column from list/array to Spark dataframe
  16. java.lang.NoClassDefFoundError: org/apache/spark/streaming/twitter/TwitterUtils$ while running TwitterPopularTags
  17. How to get ID of a map task in Spark?
  18. How to use COGROUP for large datasets
  19. Spark: produce RDD[(X, X)] of all possible combinations from RDD[X]
  20. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  21. Create new Dataframe with empty/null field values
  22. Encode an ADT / sealed trait hierarchy into Spark DataSet column
  23. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  24. Replace null values in Spark DataFrame
  25. Spark RDD default number of partitions
  26. Why does Spark RDD partition has 2GB limit for HDFS?
  27. Calculate Cosine Similarity Spark Dataframe
  28. How to print the contents of RDD?
  29. Scala-Spark Dynamically call groupby and agg with parameter values
  30. Joining two dataframes without a common column

Leave a Comment