Why is predicate pushdown not used in typed Dataset API (vs untyped DataFrame API)?

by

That’s the line in your Physical Plan you should remember to know the real difference between Dataset[T] and DataFrame (which is Dataset[Row]).

Filter <function1>.apply

I keep saying that people should stay away from the typed Dataset API and keep using the untyped DataFrame API as the Scala code becomes a black box to the optimizer in too many places. You’ve just hit one of these and think also about the deserialization of all the objects that Spark SQL keeps away from JVM to avoid GCs. Every time you touch the objects you literally ask Spark SQL to deserialize objects and load them onto JVM that puts a lot of pressure on GC (which will get triggered more often with the typed Dataset API as compared to the untyped DataFrame API).

See UDFs are Blackbox — Don’t Use Them Unless You’ve Got No Choice.

Quoting Reynold Xin after I asked the very same question on [email protected] mailing list:

The UDF is a black box so Spark can’t know what it is dealing with. There
are simple cases in which we can analyze the UDFs byte code and infer what
it is doing, but it is pretty difficult to do in general.

There is a JIRA ticket for such cases SPARK-14083 Analyze JVM bytecode and turn closures into Catalyst expressions, but as someone said (I think it was Adam B. on twitter) it’d be a kind of joke to expect it any time soon.

One big advantage of the Dataset API is the type safety, at the cost of performance due to heavy reliance on user-defined closures/lambdas. These closures are typically slower than expressions because we have more flexibility to optimize expressions (known data types, no virtual function calls, etc). In many cases, it’s actually not going to be very difficult to look into the byte code of these closures and figure out what they are trying to do. If we can understand them, then we can turn them directly into Catalyst expressions for more optimized executions.

// Let's try to find players born in 1999. 
// This will work, you have compile time safety... but it will not use predicate pushdown!!!
playersDs.filter(_.birthYear == 1999).explain()

The above code is equivalent to the following:

val someCodeSparkSQLCannotDoMuchOutOfIt = (p: Player) => p.birthYear == 1999
playersDs.filter(someCodeSparkSQLCannotDoMuchOutOfIt).explain()

someCodeSparkSQLCannotDoMuchOutOfIt is exactly where you put optimizations aside and let Spark Optimizer skip it.

More Related Contents:

  1. Spark SQL: apply aggregate functions to a list of columns
  2. DataFrame / Dataset groupBy behaviour/optimization
  3. Difference between DataFrame, Dataset, and RDD in Spark
  4. Multiple Aggregate operations on the same column of a spark dataframe
  5. DataFrame join optimization – Broadcast Hash Join
  6. How to access element of a VectorUDT column in a Spark DataFrame?
  7. Difference between DataFrame, Dataset, and RDD in Spark
  8. What is the maximum size for a broadcast object in Spark?
  9. How to loop through each row of dataFrame in pyspark
  10. Why does Spark think this is a cross / Cartesian join
  11. How do I add an persistent column of row ids to Spark DataFrame?
  12. Convert null values to empty array in Spark DataFrame
  13. How to get keys and values from MapType column in SparkSQL DataFrame
  14. Array Intersection in Spark SQL
  15. How to exclude multiple columns in Spark dataframe in Python
  16. When to use Spark DataFrame/Dataset API and when to use plain RDD?
  17. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  18. Find maximum row per group in Spark DataFrame
  19. Spark Dataframe distinguish columns with duplicated name
  20. How to export a table dataframe in PySpark to csv?
  21. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  22. How to save/insert each DStream into a permanent table
  23. How to get other columns when using Spark DataFrame groupby?
  24. PySpark: how to resample frequencies
  25. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  26. Derive multiple columns from a single column in a Spark DataFrame
  27. Filtering DataFrame using the length of a column
  28. spark createOrReplaceTempView vs createGlobalTempView
  29. How to improve broadcast Join speed with between condition in Spark
  30. Spark SQL referencing attributes of UDT

Leave a Comment