Tips and Traps¶
Bucketed column is only supported in Hive table at this time.
A Hive table can have both partition and bucket columns.
Suppose
t1andt2are 2 bucketed tables and with the number of bucketsb1andb2respecitvely. For bucket optimization to kick in when joining them:- The 2 tables must be bucketed on the same keys/columns. - Must joining on the bucket keys/columns. - `b1` is a multiple of `b2` or `b2` is a multiple of `b1`.When there are many bucketed table that might join with each other, the number of buckets need to be carefully designed so that efficient bucket join can always be leveraged.
Bucket for optimized filtering is available in Spark 2.4+. For examples, if the table
personhas a bucketed columnidwith an integer-compatible type, then the following query in Spark 2.4+ will be optimized to avoid a scan of the whole table. A few things to be aware here. First, you will still see a number of tasks close to the number of buckets in your Spark application. This is becuase the optimized job will still have to check all buckets of the table to see whether they are the right bucket corresponding toid=123. (If yes, Spark will scan all rows in the bucket to filter records. If not, the bucket will skipped to save time.) Second, the type of the value to compare must be compartible in order for Spark SQL to leverage bucket filtering. For example, if theidcolumn in thepersontable is of the BigInt type andid = 123is changed toid = "123"in the following query, Spark will have to do a full table scan (even if it sounds extremely stupid to do so).:::sql SELECT * FROM persons WHERE id = 123When you use multiple bucket columns in a Hive table, the hashing for bucket on a record is calculated based on a string concatenating values of all bucket columns. This means that to leverage bucket join or bucket filtering, all bucket columns must be used in joining conditions or filtering conditions.
Benefit of Partition Columns¶
- Spark supports partition pruning which skips scanning of non-needed partition files when filtering on partition columns. However, notice that partition columns does not help much on joining in Spark. For more discussions on this, please refer to Partition-wise joins and Apache Spark SQL .
When to Use Partition Columns¶
- Table size is big (> 50G).
- The table has low cardinality columns which are frequently used in filtering conditions.
How to Choose Partition Columns¶
Choose low cardinality columns as partition columns (since a HDFS directory will be created for each partition value combination). Generally speaking, the total number of partition combinations should be less than 50K.
The columns used frequently in filtering conditions.
Use at most 2 partition columns as each partition column creates a new layer of directory.
Bucket columns in Hive tables are similar to primary indexes in Teradata.
Benefits of Bucket Columns¶
Spark supports bucket pruning which skips scanning of non-needed bucket files when filtering on bucket columns.
Bucket join will be leveraged when the 2 joining tables are both bucketed by joining keys of the same data type and bucket numbers of the 2 tables have a times relationship (e.g., 500 vs 1000).
The number of buckets helps guide Spark engine on parallel execution level.
When to Use Bucket Columns¶
Table size is big (> 200G).
The table has high cardinality columns which are frequently used as filtering and/or joining keys.
medium size table but it is mainly used to join with a huge bucketize table, it is still beneficial to bucketize it
the sort merge join (without bucket) is slow due to shuffle not due to data skew
How to Configure Bucket Columns¶
- Choose high cardinality columns as bucket columns.
- Try to avoid data skew.
- At least 500 buckets (as a small bucket number will cause poor parallel execution).
- Sorting buckets is optional but strongly recommended.
Small Files¶
name node, each file is an object in name node, lots of small files put lots of pressure on the name node
also, compute engine also have latencies managing jobs. Lots of small files might results lots of small tasks which downgrade the performance of the Spark compute engine.
Frequent insert DML will introduce many small files in your tables' HDFS directory which will downgrade query performance and even impact HDFS name node stability.
To avoid too many small files:
- do not use "insert into tables values ..." in iterations.
- Trigger regular compact in your data processing compact table sales; # or compact table sales partition (site=0);
Analyze Tables/Columns¶
Table and column level statitics info can help the Spark compute engine to accelerate the query performance with accurate estimation. It is suggested that you populate it after your regular data processing.
analyze table sales computer statistics -- table level analyze table sales compute statistics for columns item_id -- column level analyze tables in db_name compute statistics -- analyze all tables in db_name
Misc¶
Do not use complex view definition.
Good Practices¶
- use partition/bucket columns in filtering conditions
- Spark 3 automatically handle simple situations of data skew, however, it doesn't work for complicated query especially when data skew happens in sub quries. It is suggested that you manually optimize your query for data skew issues at this time.
- cast data types to be the same for bucket joins
- consolidate UDFs -> use a sub query to invoke a UDF once
- use view to manage data access only. Avoid complex logics in a veiw definition. NEVER nest view logic in your data process flow!
- Add
explain optimizebefore your query to analyze data and recommendation on your query! - put small join table ahead (but I think Spark 3 handles this automatically now)
- A like X or A like Y -> A like any(X, Y)
Data Engineer¶
- use Parquet format, use Delta tables if you want to support update/delete, otherwise, use Hive tables
- partition/bucket on large tables and sort buckets
- reducing the number of small files
- avoid complex view definition
- data cache
- Bloom Filter Index
Tricks and Trap on DataFrame.write.partitionBy and DataFrame.write.bucketBy¶
Partition is an important concept in Spark
which affects Spark performance in many ways.
When reading a table to Spark,
the number of partitions in memory equals to the number of files on disk if each file is smaller than the block size,
otherwise,
there will be more partitions in memory than the number of files on disk.
Generally speaking,
there shouldn't be too many small files in a table as this cause too many partitions (and thus small tasks) in the Spark job.
When you write a Spark DataFrame into disk,
the number of files on disk usually equals to the number of partitions in memory
unless you use partitionBy or bucketBy.
Suppose there is a DataFrame df which has p partitions in memory
and it has a column named col which has c distinct values $v_1$, ..., $v_c$,
when you write df to disk using df.write.partitionBy(col),
each of the p partitions in memory is written to separate partitions into the c directories on disk.
This means that the final resulted number of partitions can be up to $c * p$.
This is probably not what people want in most situations,
instead,
people often want exact $c$ partitions on disk when they call df.write.partitionBy(col).
According to the above explanation on how Data.write.partitionBy works,
a simple fix is to have each partition in memory corresponding to a distinct value in the columnd df.col.
That is a repartition of the DataFrame using the col col resolves the issue.
:::python
df.repartition(col).partitionBy(col)
The above issue is not present when you DataFrame.write.bucketBy
as DataFrame.write.bucketBy works by calculating hash code.
There will always be the exact number of buckets/partitions on the disk
as you specifed when you call the function DataFrame.write.bucketBy.
import findspark
findspark.init("/opt/spark-3.0.0-bin-hadoop3.2/")
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import *
from pyspark.sql.types import StructType
spark = SparkSession.builder.appName("PySpark_Union").enableHiveSupport().getOrCreate()
df = spark.read.option("header", "true").csv("../../home/media/data/daily.csv")
df = df.select(
year("date").alias("year"), month("date").alias("month"), "date", "x", "y", "z"
).repartition(2)
df.show()
df.rdd.getNumPartitions()
df.write.mode("overwrite").partitionBy("year").parquet("part_by_year.parquet")
!ls part_by_year.parquet/
Spark support multiple levels of partition.
df.write.mode("overwrite").partitionBy("year", "month").parquet(
"part_by_year_month.parquet"
)
!ls part_by_year_month.parquet/year=2018/
spark.read.parquet("daily.parquet").rdd.getNumPartitions()
df.repartition("year").write.mode("overwrite").partitionBy("year").parquet(
"daily.parquet"
)
!ls daily.parquet/year=2018
!ls daily.parquet/year=2019
spark.read.parquet("daily.parquet").rdd.getNumPartitions()
df.write.mode("overwrite").partitionBy("year").saveAsTable("daily_hive")
spark.table("daily_hive").rdd.getNumPartitions()
df.createOrReplaceTempView("df")
spark.sql(
"""
create table daily_hive_2
using parquet
partitioned by (year) as
select * from df
"""
)
spark.table("daily_hive_2").rdd.getNumPartitions()
Filtering Optimization Leveraging Bucketed Columns¶
Spark 3¶
import findspark
findspark.init("/opt/spark-3.0.0-bin-hadoop3.2/")
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import *
from pyspark.sql.types import StructType
spark = SparkSession.builder.appName("PySpark_Union").enableHiveSupport().getOrCreate()
df = spark.read.option("header", "true").csv("../../home/media/data/daily.csv")
df = df.repartition(2)
df.show()
df.rdd.getNumPartitions()
df.write.bucketBy(10, "date").saveAsTable("daily_b2")
spark.table("daily_b2").rdd.getNumPartitions()
Notice the execution plan does leverage bucketed columns for optimization.
spark.sql(
"""
select
*
from
daily_b
where
date = "2019-01-11"
"""
).explain()
Spark 2.3¶
import findspark
findspark.init("/opt/spark-2.3.4-bin-hadoop2.7/")
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import *
from pyspark.sql.types import StructType
spark23 = (
SparkSession.builder.appName("PySpark_Union").enableHiveSupport().getOrCreate()
)
df = spark23.read.option("header", "true").csv("../../home/media/data/daily.csv")
df.show()
df.write.bucketBy(10, "date").saveAsTable("daily_b")
spark23.table("daily_b").rdd.getNumPartitions()
Notice the execution plan does not leverage bucketed columns for optimization.
spark23.sql(
"""
select
*
from
daily_b
where
date = "2019-01-11"
"""
).explain()