Vectorized execution in Spark
In prev blog post I discussed about volcano iterator and WSCG. Volacano iterator model suffered from too many virtual calls and WSCG replace multiple calls into single function call but still execution is row based. Probelm with row based execution is that it can’t utilise performance benefit from SIMD, loop unrooling etc.
Spark introduced WSCG and vectorized execution as part of project Tungsten.
Vectorization: What is Vector Processing?
Vector Processing is basically single instruction operating on one-dimensional arrays of data called vectors, compared to scalar processors, whose instructions operate on single data items as shown in the below picture:
Row vs Columnar Storage format:
In column format group of rows are batched and stored in column order <c1, c2, c3 …cn>.
I will not go into details of benefit of columnar format as it is well known.
Spark physical operators now has one more function which denotes whether operator support columnar execution.
def supportsColumnar: Boolean = false
If an operator supports columnar execution, it will be set to true (default=false).
Get Subham Singhal’s stories in your inbox
Join Medium for free to get updates from this writer.
And provides funtion for columnar execution:
final def executeColumnar(): RDD[ColumnarBatch]
Operator which supports columnar execution would implement this function and return RDD of ColumnarBatch. Its row based counter part is
final def execute(): RDD[InternalRow]: returns RDD of InternalRow
Spark provides helper function “toRowBased” to convert columnar batch to row based storage incase any operator does not support vectorized execution.
Combining WholeStageCodeGen with Columnar Execution:
Traditionally, WholeStageCodeGen was designed to optimize row-based operators. The entire pipeline of operators like projection, filtering, and aggregation is fused into a single block of Java code that operates row-by-row. WSCG does not produce code which handle columnar storage.
This is where toRowBased function becomes useful. For columnar datasource like parquet vectorized reader is used but if downstream operators does not support columnar operation, it will be converted to row based format and WSCG can be applied. WSCG and columnar execution both exist in spark but which one will be applied depends on operators getting used and their support for columnar execution.
Operator-Specific Behavior:
Operators like ProjectExec and FilterExec: These have both row-based and columnar implementations. If the columnar version is chosen, Spark will use columnar processing without applying WholeStageCodeGen.
Operators without columnar versions: For operators that don’t have columnar equivalents, Spark may fall back to row-based execution, and WholeStageCodeGen will be triggered to optimize those stages.
In summary:
- WholeStageCodeGen: Optimizes row-based operations by fusing operators into a single stage of generated code.
- Columnar Execution: Optimizes batch processing of columnar data using vectorized operators.
- Spark can use both techniques together in different stages of an execution plan, but they target different types of data processing optimizations.
