Experience Summary

Updated on 2022-06-01 GMT+08:00

View PDF

Using mapPartitions to Calculate Data by Partition

If the overhead of each record is high, for example,

rdd.map{x=>conn=getDBConn;conn.write(x.toString);conn.close}

use mapPartitions to calculate data by partition.

rdd.mapPartitions(records => conn.getDBConn;for(item <- records)
write(item.toString); conn.close)

mapPartitions can flexibly operate data. For example, to calculate the top N of a large data block, mapPartitions can be used to calculate the top N of each partition and then sort the top N of all partitions if N is a small value. Compared with calculating top N with full data, this method has a higher efficiency.

Using coalesce to Adjust the Number of Slices

Use coalesce to adjust the number of slices. The coalesce function has two parameters.

coalesce(numPartitions: Int, shuffle: Boolean = false)

If the value of shuffle is true, coalesce has the same function as repartition(numPartitions: Int). It recreates partitions using the shuffle. If the value of shuffle is set to false, partitions of a parent RDD are calculated in the same task. In this case, if the value of numPartitions is greater than the number of slices of the parent RDD, partitions are not recreated.

The coalesce operator can be used in the following scenarios:

If the previous operation involves a large number of filters, use coalesce to minimize the number of zero-loaded tasks. In coalesce (numPartitions, false), the value of numPartitions is smaller than the number of slices of the parent RDD.
Use coalesce when the number of slices entered is too large to execute.
Use coalesce when the programs are suspended in the shuffle operation because of a large number of tasks or limited Linux resources. In this case, use coalesce (numPartitions, true) to recreate partitions.

Configuring localDir

During the shuffle procedure of Spark, data needs to be written into local disks. The performance bottleneck of Spark is shuffle, and the bottleneck of shuffle is the I/O. To improve the I/O performance, you can configure multiple disks to implement concurrent data writing. If multiple disks are mounted to a node, configure a Spark localDir for each disk. This can effectively distribute shuffle files in multiple locations, improving disk I/O efficiency. The performance cannot be improved if a disk is configured with multiple directories.

Using the Collect Operation for Small Data

The collect operation does not apply to a large volume of data.

When the collect operation is performed, the executor data is sent to the driver. If the driver does not have sufficient memory, OutOfMemory occurs on the driver. Therefore, if the data volume is unknown, perform the saveAsTextFile operation to write data into HDFS. If the data volume is known and the driver has sufficient memory, perform the collect operation.

Using reduceByKey

The reduceByKey operator implements local aggregation on the Map side, which offers a smooth shuffle procedure. The groupByKey operator, however, does not perform aggregation on the Map side. Therefore, use reduceByKey if possible to avoid implementation modes like groupByKey().map(x=>(x._1,x._2.size)).

Broadcasting Map Instead of Arrays

If a table query is required for each record of data that is broadcast from the driver side, broadcast the data in the set/map structure instead of Iterator. The query speed of the set/map structure is approximately O(1), while that of Iterator is O(n).

Avoiding Data Skew

If data skew occurs (certain data volume is extremely large), the execution time of tasks is inconsistent even if no GC is performed.

Redefine the keys. Use keys of smaller granularity to optimize the task size.
Modify the DOP.

Optimizing the Data Structure

Store data by column. In this way, only the required columns are scanned when data is read.
When using Hash Shuffle, set spark.shuffle.consolidateFiles to true to combine intermediate files of shuffle, minimize the number of shuffle files and file I/O operations, and improve performance. The number of final files is the number of reduce tasks.

Parent topic: Spark Core Tuning

Previous topic: Designing a Direction Acyclic Graph (DAG)

Next topic: SQL and DataFrame Tuning