Partitioning hints allow you to suggest a partitioning strategy that Databricks should follow. COALESCE, REPARTITION, and REPARTITION_BY_RANGE hints are supported and are equivalent to coalesce, repartition, and repartitionByRange Dataset APIs, respectively. These hints give you a way to tune performance and control the number of output files. When multiple partitioning hints are specified, multiple nodes are inserted into the logical plan, but the leftmost hint is picked by the optimizer.
Reduce the number of partitions to the specified number of partitions. It takes a partition number as a parameter.
Repartition to the specified number of partitions using the specified partitioning expressions. It takes a partition number, column names, or both as parameters.
Repartition to the specified number of partitions using the specified partitioning expressions. It takes column names an optional partition number as parameters
The REBALANCE hint can be used to rebalance the query result output partitions, so that every partition is of a reasonable size (not too small and not too big). It can take column names as parameters, and try its best to partition the query result by these columns. This is a best-effort: if there are skews, Spark will split the skewed partitions, to make these partitions not too big. This hint is useful when you need to write the result of this query to a table, to avoid too small/big files. This hint is ignored if AQE is not enabled.
> SELECT /*+ COALESCE(3) */ * FROM t;
> SELECT /*+ REPARTITION(3) */ * FROM t;
> SELECT /*+ REPARTITION(c) */ * FROM t;
> SELECT /*+ REPARTITION(3, c) */ * FROM t;
> SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t;
> SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t;
> SELECT /*+ REBALANCE */ * FROM t;
> SELECT /*+ REBALANCE(c) */ * FROM t;
-- When a column name has been occluded by an alias you must refere to it by the alias name.
> SELECT /*+ REBALANCE(d) */ * FROM t AS s(d);
-- multiple partitioning hints
> EXPLAIN EXTENDED SELECT /*+ REPARTITION(100), COALESCE(500), REPARTITION_BY_RANGE(3, c) */ * FROM t;
== Parsed Logical Plan ==
'UnresolvedHint REPARTITION, [100]
+- 'UnresolvedHint COALESCE, [500]
+- 'UnresolvedHint REPARTITION_BY_RANGE, [3, 'c]
+- 'Project [*]
+- 'UnresolvedRelation [t]
== Analyzed Logical Plan ==
name: string, c: int
Repartition 100, true
+- Repartition 500, false
+- RepartitionByExpression [c#30 ASC NULLS FIRST], 3
+- Project [name#29, c#30]
+- SubqueryAlias spark_catalog.default.t
+- Relation[name#29,c#30] parquet
== Optimized Logical Plan ==
Repartition 100, true
+- Relation[name#29,c#30] parquet
== Physical Plan ==
Exchange RoundRobinPartitioning(100), false, [id=#121]
+- *(1) ColumnarToRow
+- FileScan parquet default.t[name#29,c#30] Batched: true, DataFilters: [], Format: Parquet,
Location: CatalogFileIndex[file:/spark/spark-warehouse/t], PartitionFilters: [],
PushedFilters: [], ReadSchema: struct<name:string>
When you write PySpark DataFrame to disk by calling partitionBy(), PySpark splits the records based on the partition column and stores each partition data into a sub-directory.
# partitionBy() Example
df.write.option("header",True) \
.partitionBy("state") \
.mode("overwrite") \
.csv("/tmp/zipcodes-state")
My data has six states; hence, it created six directories on disk. The name of the sub-directory would be the partition column and its value (partition column=value).
When writing the data as partitions, PySpark eliminates the partition column on the data file and adds the partition column and value to the folder name, saving some storage space. To validate this, open any partition file in a text editor and check.