As data sizes have grown in enterprises across industries, Apache Parquet has become a prominent format for storing data. Apache Parquet is a columnar storage format designed for efficient data processing at scale. By organizing data by columns rather than rows, Parquet enables high-performance querying and analysis, as it can read only the necessary columns for a query instead of scanning entire rows of data. Parquet’s efficient data layout has made it a popular choice in the modern analytics ecosystem, specifically with Apache Spark workloads.
The RAPIDS Accelerator for Apache Spark built on cuDF supports Parquet as a data format for reading and writing data in an accelerated manner on GPUs. For many large-scale Spark workloads where data input sizes are in terabytes, having efficient Parquet scans is critical for achieving good runtime performance.
In this post, we discuss how to mitigate the occupancy limitations caused by higher register usages and share benchmark results.
Apache Parquet data format
The Parquet file format enables data to be stored in columnar format using column chunks assembled into row groups. The metadata is distinct from the data to enable the splitting of columns into multiple files as necessary (Figure 1).
The Parquet format supports a variety of data types. Metadata specifies how these types should be interpreted, which enables these types to represent more complex logical types, such as time stamps, strings, decimals, and so on.
You can also use metadata to specify more complex structures, such as nested types and lists. The data can be encoded in a variety of different formats, such as plain values, dictionaries, run-length encoding, bit-packing, and more.
- BOOLEAN: 1 bit boolean- INT32: 32 bit signed ints- INT64: 64 bit signed ints- INT96: 96 bit signed ints- FLOAT: IEEE 32-bit floating point values- DOUBLE: IEEE 64-bit floating point values- BYTE_ARRAY: arbitrarily long byte arrays- FIXED_LEN_BYTE_ARRAY: fixed length byte arrays |
Parquet on GPU occupancy limitations
Before the RAPIDS Accelerator for Apache Spark, the previous implementation of Parquet scans was a monolithic cuDF kernel that supported all Parquet column types in a single set of processing code.
As adoption of Spark on GPUs increased for customers with Parquet data, more time was invested in understanding the performance characteristics of Parquet scans given the critical component of performance that it represented. There are several general resources that factor into how efficiently a kernel can operate:
- Streaming microprocessors (SMs): Main processing unit of the GPU, which is responsible for executing computing tasks.
- Shared memory: GPU on-chip memory that is allocated per thread-block, so that all threads in the same block can access the same shared memory.
- Registers: Fast on-chip GPU memory that stores information used by a single thread for the compute operations that are executed by the SMs.
As we analyzed Parquet scans, we observed that the overall GPU occupancy was lower than desired due to encountering register limits. The register usage is determined by how the CUDA compiler generates code based on the kernel logic plus data management.
For the Parquet monolithic kernel, the complexity to support all column types created a large, complex kernel with high shared memory and register usage. While a single monolithic kernel may have consolidated code together, its complexity constrained the possible types of optimizations and resulted in performance limitations at scale.
Figure 2 represents the Parquet data processing loop on the GPU. Each of the blocks is a large and complex amount of kernel code, which may have its own shared memory requirements. Many of the blocks are type-dependent, which leads to a bloated kernel loaded into memory.
Specifically, one of the limitations was how Parquet blocks were decoded inside warps. There was a serial dependency for warps to wait for previously ordered warps to complete before processing their blocks of data. This enabled different parts of the decoding process to happen across different warps, but created an inefficient dependency for work to be accomplished on the GPU.
Moving to a block-wide decoding algorithm was critical for performance but would have increased the register count and limited occupancy even further due to its added data sharing and synchronization complexity.
Parquet microkernels in cuDF
To mitigate the occupancy limitations caused by higher register usages, we tried an initial idea of a smaller kernel for preprocessing list type data in Parquet. We separated a piece of code out of the monolithic kernel into a self-contained kernel and the results were impressive. The overall benchmarks showed faster runtimes and the GPU traces demonstrated improved occupancy.
After that, we tried the same approach for different column types. The microkernels for various types used C++ templates to reuse capabilities. This simplified the code for maintenance and debugging per type.
The Parquet microkernel approach takes advantage of compile time optimization to only go through the necessary code paths for processing a given type. Instead of one monolithic kernel containing all possible code paths, you can generate many individual microkernels containing just the code needed for that path.
This can be done using if constexpr at compile time so that the code reads normally, but doesn’t include any code paths that will never be taken for a particular combination of data attributes (strings or fixed-width, lists or no-lists, and so on.).
Here is a simple example for processing columns of fixed width type. You can see that the majority of processing isn’t needed and is skipped in the new microkernel approach. Only the data copying is needed for this type.
To solve the inter-warp bottleneck, the new microkernels enabled entire blocks to be processed at each step so warps could independently process data more efficiently. This was especially significant for strings to enable a full block of 128 threads on the GPU to copy strings where the previous implementation only used one warp to copy strings.
We ran local benchmarks using a NVIDIA RTX A5000 GPU 24GB with compressed Parquet Snappy 512 MB data preloaded in device buffers. For testing chunked reads, we read 500-KB chunks at a time. The test data included some variations:
- Cardinality 0 and 1000
- Run length 1 and 32
- 1% nulls
- Adaptive dictionary usage to be used if data was repeated
Figure 5 shows the results in terms of throughput improvement across Parquet column types with the new microkernel approach on the GPU.
An optimization to the chunked reading of list columns also improved throughput by 117% for 500-KB reads.
Getting started with Apache Spark on GPUs
Parquet is a key data format in wide adoption for large data processing. The GPU can accelerate scanning Parquet data in Apache Spark through the use of optimized microkernels in cuDF.
Enterprises can take advantage of the RAPIDS Accelerator for Apache Spark to seamlessly migrate Apache Spark workloads to NVIDIA GPUs. RAPIDS Accelerator for Apache Spark leverages GPUs to accelerate processing by combining the power of the RAPIDS cuDF library and the scale of the Spark distributed computing framework. Run existing Apache Spark applications on GPUs with no code changes by launching Spark with the RAPIDS Accelerator for the Apache Spark plugin JAR file.
Get hands-on with Parquet scan processing and the RAPIDS Accelerator for Apache Spark with the Spark RAPIDS Parquet acceleration Colab notebook.
SAGECONNECT TECHNOLOGY GROUP CO., LIMITED
Sageconnect is a reliable provider of IT network system integration and solutions for global enterprise customers.
