The workload for this test is a simple query reading from a partitioned table in Apache Parquet format. The test table used in this post is WEB_SALES, a partitioned table of ~68 GB, part of the TPCDS benchmark (scale 1500GB). The setup instructions for the TPCDS database on Spark can be found at: "Spark SQL Performance Tests".

The tests have been performed using Spark 2.2.0, running in local mode. The test machine is a dual socket server with 2 Intel(R) Xeon(R) CPU E5-2630 v4 @ 2.20GHz (10 cores per CPU, 20 cores in total). The test server has 512 GB of RAM (16 DIMMs DDR4 of 32GB each). It is important for the tests/labs described here to have enough memory to cache the test table and to provide enough RAM for the executors (128 GB of RAM in the test machine is probably the minimum if you want to reproduce the results at the scale described in this post). The storage of the test machine is provided by just two local disks, this is undersized compared to but it is not matter much for the results of the tests, as the test workload is CPU-bound. The server used for the tests described here was installed with Linux, RHEL 6.8.

Some of the labs described in post make reference to Spark task metrics, in particular collected using a tool called sparkMeasure, described in this post. Also relevant are additional examples and lab setup instructions as discussed in "Diving into Spark and Parquet Workloads, by Example".

 

 

The first observation that got me started with the investigations in this post is that reading Parquet files with Spark is often a CPU-intensive operation. A simple test to realize this is by reading a test table using a Spark job running with just one task/core and measure the workload using Spark. This is the subject of the first experiment:

 

Lab 1: Experiment with reading a table from Parquet and measure the workload

The idea for this experiment is to run a Spark job to read a test table stored in Parquet format (see lab setup details above). The job just scans through the table partitions with a filter condition that is written in such a way that it is never true for the given test data, this has the effect of generating I/O to read the entire table, producing a minimal amount of overhead for processing and returning the result set (which will be empty). From more details of how this "trick" works, see the post "Diving into Spark and Parquet Workloads, by Example".

One more prerequisite, before starting the test, is to clean the file system cache, this ensure reproducibility of the results if you want to run it the test multiple times:

 

Spark Context default degree of parallelism = 1

Aggregated Spark stage metrics:

numStages => 1

sum(numTasks) => 787

elapsedTime => 465430 (7.8 min)

sum(stageDuration) => 465430 (7.8 min)

sum(executorRunTime) => 463966 (7.7 min)

sum(executorCpuTime) => 325077 (5.4 min)

sum(executorDeserializeTime) => 905 (0.9 s)

sum(executorDeserializeCpuTime) => 917 (0.9 s)

sum(resultSerializationTime) => 14 (14 ms)

sum(jvmGCTime) => 5793 (6 s)

sum(shuffleFetchWaitTime) => 0 (0 ms)

sum(shuffleWriteTime) => 0 (0 ms)

max(resultSize) => 1067344 (1042.0 KB)

sum(numUpdatedBlockStatuses) => 0

sum(diskBytesSpilled) => 0 (0 Bytes)

sum(memoryBytesSpilled) => 0 (0 Bytes)

max(peakExecutionMemory) => 0

sum(recordsRead) => 1079971165

sum(bytesRead) => 72942144585 (67.9 GB)

sum(recordsWritten) => 0

sum(bytesWritten) => 0 (0 Bytes)

sum(shuffleTotalBytesRead) => 0 (0 Bytes)

sum(shuffleTotalBlocksFetched) => 0

sum(shuffleLocalBlocksFetched) => 0

sum(shuffleRemoteBlocksFetched) => 0

sum(shuffleBytesWritten) => 0 (0 Bytes)

sum(shuffleRecordsWritten) => 0

 

What you can learn from this test and the collected metrics values:

• The workload consists in reading one partitioned table, the web_sales table part of the TPCDS benchmark. The table size is ~68 GB (it has ~1 billion rows), generated as described above in "Lab setup".
• From the metrics you can see that the query ran in about 8 minutes, with only one active task at a time. Note also that "executor time" = "elapsed time" in this example.
• Most of the execution time was spent on CPU. Other metrics about time measures (serialization/deserialization, garbage collection, shuffle), although of importance in general for Spark workloads, can be ignored in this example as their values are very close to zero.
• The difference between between the elapsed time and the CPU time can be interpreted as the I/O read time (more on this later in this post). The estimated value of the time spent on I/O amounts to ~ 30% of the total execution time. Further in this post you can see how to directly measure I/O time and confirm this finding.
• We conclude that the workload is mostly CPU-bound 

 

Lab 2: focus on the CPU time of the workload

...

elapsedTime => 316575 (5.3 min)

sum(stageDuration) => 316575 (5.3 min)

sum(executorRunTime) => 315270 (5.3 min)

sum(executorCpuTime) => 310535 (5.2 min)

sum(executorDeserializeTime) => 867 (0.9 s)

sum(executorDeserializeCpuTime) => 874 (0.9 s)

sum(resultSerializationTime) => 13 (13 ms)

sum(jvmGCTime) => 4961 (5 s)

 

Lab 3: Measure Spark workload with OS tools

This is about using OS tools to measure systems resource utilization for the test workload. By contrast, the Spark workload metrics you have seen so far in the results of experiments 1 and 2 are measured by the Spark executor itself. If you are interested in the details, you can see the code of how this is done in Executor.scala, for example, the CPU time metric is collected using ThreadMXBean and the method getCurrentThreadCpuTime(), Garbage collection time comes from the GarbageCollectorMXBean and the method getCollectionTime().

This experiment is about measuring CPU usage with OS tools and comparing the results with the metrics reported by Spark Measure (as in the tests of Lab 1 and 2). There are multiple tools you can use to measure process CPU usage in Linux, ultimately with instrumentation coming from /proc filesystem. This is an example of using the Linux comman ps:

 

# use this to find pid_Spark_JVM

$ jps | grep SparkSubmit 

# measure cpu time 

$ ps -efo cputime -p <pid_Spark_JVM>

... run test query in the spark-shell as in Lab 2 and run again:

$ ps -efo cputime -p <pid_Spark_JVM>

Dynamic tracing: Measuring the latency of I/O calls is a good task for dynamic tracing tools, like SystemTap, bcc/BPF, DTrace. Earlier in this post, I commented that in Lab 1 the query execution spent 70% of its time on CPU and 30% of the time was spent on I/O (reading from Parquet files): the CPU time is a direct measurement reported via Spark metrics, the time spent on I/O is inferred from subtracting the total executor time from CPU time.

 

Lab 4: Drilling down with stack profiling and flame graphs

Stack profiling and on-CPU Flame Graph visualization are useful techniques and tools for investigating CPU workloads, see also Brendan Gregg's page on Flame Graphs.

Stack profiling helps in understanding and drilling-down on "hot code":

you can find which parts of the code spend considerable amount of time on CPU and this provides insights for troubleshooting.

Flame graph visualization of the stack profiles brings additional value, besides being an appealing interface, it provides context about the running code. In particular flame graphs provide information on the parent functions, this can be used to match the names of the functions consuming CPU resources and their parents with the application architecture/code. There are other methods to represent stack profiles, including reporting a summary or representing it in a tree view. Nitsan Wakart in "Exploring Java Perf Flamegraphs" explains the advantages of using flame graphs and compares a few of the available tools for collecting stack profiles for the JVM. See also notes and examples of how to use tools for stack sampling and flame graph visualization for Spark at this link.

 

What you can learn from the flame graph:

• The flame graph shows the stack traces of the parts of the code that take most of the time. Code paths for execution of Spark data sources and "Vectorized Parquet Reader" are clearly occupying most of the trace samples (i.e. are taking most of the CPU time). 
• You can see also that the execution is under "whole stage code generation", another important feature of Spark 2.x (see also the discussion in the blog entry "Apache Spark 2.0 Performance Improvements Investigated With Flame Graphs".

There are many interesting details to explore in the flame graph, for this it is best to use SVG version with the zoom feature (click on the graph to zoom) and the search feature (use Control-F). See also the exercise at the end of Lab 5 and the memory heap profile in Lab 6.

 

val df = spark.table("web_Sales")

df.write

  .partitionBy("ws_sold_date_sk")

  .option("compression","none")

  .parquet("<path>/web_sales_compress")

 

The next step is running a SQL statement to read the entire table (similarly to previous Labs, this time, however, using the non-compressed table data). As in the case of Lab 2 above the test is done after caching the table in the file system cache, which happens naturally on a system with enough memory after running the query a couple of times. The metrics collected with sparkMeasure when running the test are:

 

Lab 6: Measuring the effect of garbage collection

Part 2: Scaling up the workload

• The test workload, reading from a Parquet partitioned table, scales up on a 20-core server.
• The workload is instruction-bound up to 10 concurrent tasks, at higher load it is limited by CPU-to-memory bandwidth.
• CPU utilization metrics can be misleading, drill down by measuring instruction per cycle (IPC) and memory throughput. Beware of hyper-threading.

 

Lab 8: Measuring elapsed time at scale

In the experiments described so far you have seen how the test workload (reading from a Parquet partitioned table) runs on Spark using a single task and how to measure it in some simple and controlled cases. Distributed data processing and Spark are all about running tasks concurrently, this is the subject of the the rest of this post.

The test machine I used has 20 cores (2 sockets, with 10 cores each, see also Lab setup earlier in this post). The results of running the workload with increasing number of concurrent tasks is summarized in the following table:

Concurrent Tasks (N#)	Elapsed time (millisec)
1	315816
2	151787
3	102726
4	76959
...	...
19	20638
20	19917

 

Key takeaway point from the table:

 

• As the number of concurrent tasks deployed increases, the job duration decreases, as expected. 
• The measurements show that elapsed time drops from about 5 minutes, in the case of no parallelism, down to 20 seconds, when 20 CPU cores are fully utilized. 
• The next question is to quantify how well the workload scales on multiple CPU cores, which is the subject of the next Lab.

 

Note on the measurement method: measurements in the top scale of the number of concurrent tasks shows an important amount of time spent on garbage collection (jvmGCTime) in their first execution. In subsequent executions the garbage collection time is gradually reduced to just a few seconds. The measurement reported here are collected after a few "warm up runs" of the query (typically around 3 "warm up" runs before measuring). For similar reason the parameter --driver-memory is "oversize" to a high value (typically 32g * num_executors).

 

Lab 9: How well does the workload scale?

p=N# executors	Elapsed time (s)	Speedup(p)
1	311302	1.00
2	149673	2.08
3	100194	3.11
4	75547	4.12
5	60931	5.11
6	51342	6.06
7	44165	7.05
8	39613	7.86
9	35450	8.78
10	32061	9.71
11	29436	10.58
12	27735	11.22
13	25764	12.08
14	24021	12.96
15	23217	13.41
16	22092	14.09
17	21016	14.81
18	20647	15.08
19	19999	15.57
20	19989	15.57

 

Lessons learned:

• The speedup grows almost linearly, at least up to about 10 concurrent tasks. 
• This can be interpreted as a sign that the workload scales and does not encounter bottlenecks at least up to the measured scale. 
• At higher load you can see that the speedup reaches a saturation zone. Drilling down on this is the subject of the next Lab.

 

Lab 10: CPU utilization, instructions, cycles and memory at scale

CPU utilization is a metric that does not reveal many insights about the nature of the workload. Measuring CPU instructions and cycles helps in understanding if the workload is instruction-bound or memory-bound. Also measuring the throughput between CPU and memory is important for data-intensive workloads as the one tested here.

The following graph reports the measurement of the CPU-to-memory throughput for the test workloads measured using Intel's pcm (see also this link for more info).

 

 

CPU-to-memory traffic measurement at high load - Here below some key metrics measured in the test with 20 concurrent tasks executing the test workload:

• The measured throughput of data transfer between CPU and memory is about 84 GB/s (55% of which is writes and 45% reads). 
• Another important metric is the utilization of the QPI links, there are 4 links in total in the system (2 links per socket). The measured QPI utilization is ~65% for the test with 20 concurrent processes.

 

Comparing with benchmarks - The question I would like to answer is: I have measured 84 GB/s of CPU-to-memory throughput, is that a high value for my test system? To approach the problem I have used a couple of benchmark tools to generate high load on the system: John D. McCalpin's stream memory test and Intel memory latency latency checker, see further details on those tools at this link. The test system during the stress tests has reached up to 100 GB/s of CPU-to-memory throughput with the benchmark tools. By comparing the benchmark values with the measured throughput of 84 GB/s I am tempted to conclude that the test workload (reading a Parquet table) stresses the system towards saturation levels of the CPU-to-memory channel at high load (with 20 cores busy executing tasks).

 

IPC measurement under load - Another measure hinting that the CPUs are spending an increasing amount of time in stalled cycles at high load, comes from the measurement of instruction per cycle (IPC). The number of instructions for the test workload (reading 68 GB of a test Parquet table) stays constant at about 1.3E12 instructions at various load levels (this number can be measured with perf for example). However, the number of cycles needed to process the workload increases with load. Measurements on a the test system are reported in the following figure:

 

 

   

Lessons learned: 

 

In the previous paragraphs you have seen that when the number of concurrent tasks goes above the number of available CPU cores in the system, the throughput of the test workload does not increase, even though the OS reports an increasing utilization of the CPU threads.

What happens when the number of concurrent tasks keeps increasing and goes above the number of available CPU threads in the system?

This is another case of running on a system with high load and it is a valid case, possibly because you are running on a system shared with other users creating additional load. Here is an example, when running the workload with 60 concurrent tasks (the test system has 40 CPU threads):

Metric	20 concurrent tasks	40 concurrent tasks	60 concurrent tasks
Elapsed time	20 s	23 s	23 s
Executor run time	392 s	892 s	1354 s
Executor CPU Time	376 s	849 s	872 s
CPU-memory data volume	1.6 TB	2.2 TB	2.2 TB
CPU-memory throughput	85 GB/s	90 GB/s	90 GB/s
IPC	1.42	0.66	0.63
QPI utilization	65%	80%	80%

 

Key points: 

• At high load the job latency stays constant with increasing load, roughly to the value reached when running with 20 concurrent tasks (that is about 20 seconds, 20 concurrent tasks corresponds to one task per core in the test system).
• The amount of utilized CPU seconds does not increase above the value measured at 40 concurrent tasks. This is explained as the OS can schedule work on 40 "logical CPUs" corresponding to 40 possible concurrent execution threads in the CPUs using hyper-threading.
• When the number of concurrent tasks is higher than the number of available CPU threads (40) the measured CPU utilization time does not increase, while the sum of the elapsed time on the executor processes keeps increasing. The additional executor time measured accounts for time spent waiting in OS scheduler queues.
• Gotcha: The amount of useful work that the system does when executing the test workload at 20, 40 or 60 concurrent tasks is very similar. In all cases the workload is CPU-bound, with high throughput to memory. The measured metrics values CPU time and executor run time can be misleading at high load for the reasons just explained. It is useful to bear in mind this type of examples when troubleshooting workloads on shared/multi-tenant systems, where additional load on the system can come from sources external to your jobs.

 

CPU-bound workloads  are very important in data processing. In particular modern hardware has typically abundance of cores and memory. Technologies like Spark, Parquet, Hadoop, make data processing at scale easier and cheaper. Techniques and tools for measuring and understanding data processing workloads, including their CPU and memory footprint, are important to get the best value out of your software and hardware investment.

Reading from Parquet is a CPU-bound workload examined in this post. Some of the lessons learned investigating the test workloads are:

 

• Spark metrics can be used to understand resource usage and workload behavior, with some limitations.
• Use OS-tools to measure CPU, network, disk. Dynamic tracing can drill down on  latency. Stack profiling and flame graphs are useful to drill down on hot parts of the code
• CPU utilization does not always give the full picture: use hardware counters to drill down, including instructions and cycles (IPC) and CPU-to-memory throughput. 
• Measure your workload, speedup, IPC, CPU utilization, as function of increasing number of concurrent tasks to understand its scalability profile and bottlenecks.
• Watch out for common pitfalls when measuring CPU metrics on multi-tenant systems, in particular be aware if Hyper-threading is used.

Relevant articles and presentations:

 

- Brendan Gregg's blog post CPU Utilization is Wrong

- The presentation: A Crash Course in Modern Hardware by Cliff Click

- Brendan Gregg's page on Flame Graphs

- Nitsan Wakart: Using FlameGraphs To Illuminate The JVM by , Exploring Java Perf Flamegraphs

- For related topics in the Oracle context, see Tanel Poders's blog post "RAM is the new disk"

- Intel 64 and IA-32 Architectures, Optimization Reference Manual

- Intel Tuning Guides and Performance Analysis Papers

 

Acknowledgements

This  work has been done in the context of CERN Hadoop and Spark service and has profited from the collaboration of several of my colleagues at CERN.