Predicate pushdown is a technique used in Spark to filter data as early as possible in the query execution process, in order to minimize the amount of data that needs to be shuffled and processed. It allows Spark to push down filtering conditions (predicates) to the storage layer, where the data is located. Which means instead of bringing all the data into the Spark cluster first and then applying the filtering conditions.
Enabling predicate pushdown in Spark can significantly improve the performance of queries that filter large amounts of data.
In Pyspark, predicate pushdown can be enabled by setting the
spark.sql.hive.convertMetastoreParquetand
spark.sql.hive.metastorePartitionPruningIts configuration properties need t set to to true.
Sample code:
from pyspark import SparkConf, SparkContext
conf = SparkConf().setAppName("MyApp").setMaster("local")
conf.set("spark.sql.hive.convertMetastoreParquet", "true")
conf.set("spark.sql.hive.metastorePartitionPruning", "true")
sc = SparkContext(conf=conf)
You can also enable predicate pushdown while creating a Dataframe using the .filter() method in the following way:
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("pushdown").enableHiveSupport().getOrCreate()
df = spark.table("table_name")
df.filter("column_name = 'some value'").count()
It’s worth noting that for this technique to work, the data must be stored in a format that supports predicate pushdown, such as Parquet or ORC. Additionally, the optimization only works when the filter conditions are expressed in terms of the columns of the table, not on the result of an expression.
It is also worth noting that when using Hive metastore, partition pruning should be also enabled by setting spark.sql.hive.metastorePartitionPruning to true in order to push down the filtering conditions to the storage layer.
Spark important urls to refer
