Spark Memory Management

Exercises for Spark Internals.

Goal of this exercise is to learn the following

• • How to optimize the shuffle using local aggregation

  • Control parallelism by controlling partitions of RDD

  • How to optimize caching

Here is the test dataset. Download, unzip and save it in "stocks" folder in your HDFS home directory.

Controlling Partitions of RDD

• 1. Identify the narrow and wide operations in the following statement and from spark Web UI find the number of stages and the number of tasks for a job for the following

val stocks = sc.textFile("stocks")

stocks

.repartition(10)

.filter(!_.startsWith("date"))

.map(_.split(","))

.map(tokens => (tokens(7), tokens(5).toDouble))

.groupByKey()

.mapValues(values => values.sum / values.size)

.coalesce(1)

.count

• 1. Control the number of partitions while reading data from source

sc.textFile("stocks").partitions.size // Expected value is 2 as min number of spark partitions for HDFS datasource is 2.

sc.textFile("stocks", 7).partitions.size // Expected value 7

• 1. Verify that the no of partitions of shuffled rdd is same as that of parent when num partitions argument is not passed in wide operation and default level of parallelism is not set.

val stocks = sc.textFile("stocks")

stocks

.repartition(11)

.filter(!_.startsWith("date"))

.map(_.split(","))

.map(tokens => (tokens(7), tokens(5).toDouble))

.groupByKey()

.partitions.size // Expected value is 11 since the parent RDD before groupByKey operation has 11 partitions created through repartitioning.

• 1. Control the number of partitions using numPartitions argument of the wide operation

val stocks = sc.textFile("stocks")

stocks

.filter(!_.startsWith("date"))

.map(_.split(","))

.map(tokens => (tokens(7), tokens(5).toDouble))

.groupByKey(18)

.partitions.size // Expected value is 18

• 1. Control the number of partitions using default level of parallelism property

Launch spark with default parallelism as 13

$ spark-shell --conf spark.default.parallelism=13 --verbose

val stocks = sc.textFile("stocks", 7)

stocks

.filter(!_.startsWith("date"))

.map(_.split(","))

.map(tokens => (tokens(7), tokens(5).toDouble))

.groupByKey()

.partitions.size // Expected value is 13

• 1. Control the number of partitions using re-partition and coalesce function. Note: Use repartition to increase the number of partitions and use coalesce to reduce the number of partitions.

val stocks = sc.textFile("stocks")

stocks.repartition(12).partitions.size //Expected value is 12

stocks.coalesce(3).partitions.size // Expected value is 3

Solve the same problem using reduceByKey function.

val stocks = sc.textFile(“stocks”)

stocks

.filter(!_.startsWith("date"))

.map(_.split(","))

.map(tokens => (tokens(7), tokens(5).toDouble))

.reduceByKey((x, y) => x + y)

.collect

Produces shuffle write/read: 12.7 KB.

Conclusion, reduceByKey has drastically reduced the amount of network shuffle.

Caching Behaviour

Launch spark-shell with 2GB of drive memory.

$ spark-shell --verbose --driver-memory 2g

Load stocks dataset and cache the dataset in memory. Run a full scan operation using count and capture the size of the RDD in memory.

scala> import org.apache.spark.storage.StorageLevel

scala> val stocks = sc.textFile(“stocks”)

scala> stocks.cache().count // Same as stocks.persist(StorageLevel.MEMORY_ONLY)

Capture the size of the RDD in memory. See the screenshot below about where to find it.