layout: global title: Apache Arrow Cache Format displayTitle: Apache Arrow Cache Format license: | Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Apache Spark supports using Apache Arrow as an alternative cache format for in-memory Dataset caching. This format provides improved performance for certain workloads, especially when working with columnar data sources like Parquet and ORC.
The Arrow cache format offers several advantages over the default cache format:
Note that the cached bytes are an internal format (a schema-less Arrow RecordBatch payload), not a complete Arrow IPC stream, so they are not directly readable by external Arrow tooling.
Note: Spark‘s built-in Parquet/ORC readers use internal column vectors (OnHeapColumnVector/OffHeapColumnVector), not Arrow format, so they don’t benefit from zero-copy optimization.
spark.sql.cache.serializer is a static SQL configuration, so it must be set when the SparkSession is built and cannot be changed on a running session (spark.conf.set rejects static keys with CANNOT_MODIFY_CONFIG):
val spark = SparkSession.builder() .appName("MyApp") .config("spark.sql.cache.serializer", "org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer") .getOrCreate()
Note: This config selects the cache serializer for the whole session; once set, this serializer handles every cached relation. There is no automatic per-relation fallback to another cache serializer based on the data types involved (see Supported Data Types for how unsupported types are handled). The chosen serializer is also cached process-wide on first use, so switching cache formats within a JVM that has already materialized a cache requires a fresh JVM (see Migration from Default Cache).
Once configured, use cache operations as normal:
// Cache a DataFrame val df = spark.read.parquet("data.parquet") df.cache() // Use cached data df.filter("age > 30").count() // Uncache when done df.unpersist()
Arrow cache supports multiple compression codecs. Configure compression with:
spark.conf.set("spark.sql.execution.arrow.compression.codec", "zstd")
Available options:
none - No compression (fastest, largest size, default)zstd - Zstandard compression (best compression; tunable level)lz4 - LZ4 compression. Not recommended: Arrow's Java LZ4 codec is implemented with the pure-Java Commons Compress framed LZ4 streams and is much slower than zstdFor zstd, you can also configure the compression level. Positive values (up to 22) give better compression but slower speed; negative values give ultra-fast compression with lower ratios:
spark.conf.set("spark.sql.execution.arrow.compression.zstd.level", "3") // Default: 3
Enable vectorized reading for better performance with primitive types:
spark.conf.set("spark.sql.inMemoryColumnarStorage.enableVectorizedReader", "true")
When enabled, cached data is read as columnar batches instead of rows, which can significantly improve performance for columnar operations.
In our benchmarks, the Arrow cache format performs best on the following workloads. Actual results depend on data types, compression settings, and hardware, and the default cache format can be faster in some cases (for example, with higher compression levels):
ArrowColumnVector format, enabling true zero-copy extraction and the largest gains.The numbers below are illustrative results from one run on an Apple M4 Max (OpenJDK 21.0.8) and will vary with hardware, JDK, and compression settings. They are not a guarantee. For the authoritative, regularly regenerated numbers, see sql/core/benchmarks/ArrowCacheBenchmark-jdk21-results.txt and the ArrowCacheBenchmark suite.
| Workload | Default Cache | Arrow Cache | Speedup |
|---|---|---|---|
| Write + Read (5M rows, 3 primitive columns) | 153.7 ns/row | 74.2 ns/row | ~2X faster |
| Cache then filter (5M rows) | 100.1 ns/row | 70.8 ns/row | ~1.4X faster |
| Columnar input from Parquet (2M rows, 3 primitive columns) | 195.3 ns/row | 113.1 ns/row | ~1.7X faster |
| Re-cache with zero-copy (2M rows, 2 columns) | 123.3 ns/row | 38.5 ns/row | ~3.2X faster |
Notes:
OnHeapColumnVector/OffHeapColumnVector rather than ArrowColumnVector, due to Arrow’s efficient batch processingdf.drop("column")), the remaining columns preserve their ArrowColumnVector format, enabling true zero-copy extraction and achieving the best performanceArrowColumnVector (e.g., Python Arrow sources, re-caching Arrow cached data with column projection)Arrow cache supports the following data types:
TIMESTAMP_NTZ(p) / TIMESTAMP(p) with p in 7..9)Nanosecond-precision timestamps and CalendarIntervalType are stored in lossless internal Arrow representations (structs of the types' own components) rather than the standard Arrow interchange encodings, so their full Spark value domains round-trip through the cache -- the same domains the default cache serializer supports. The standard interchange encodings pack these values into a single int64 of nanoseconds and therefore cover only a reduced range (roughly years 1677-2262 for timestamps, +/-292 years of microseconds for intervals); that limitation applies to Arrow interchange paths such as toPandas, not to this cache.
Arrow cache covers every type the default cache serializer supports, plus some it does not (for example geometry and geography). Types that Arrow cannot represent (such as ObjectType) are not silently dropped or routed to a different cache serializer: there is no per-type fallback, because the cache serializer is chosen once via the static spark.sql.cache.serializer configuration and then handles every cached relation. Attempting to cache an unsupported type fails with an UNSUPPORTED_DATATYPE error when the cache is materialized.
Arrow cache automatically collects min/max statistics for the following types:
Other types (Binary, Variant, calendar intervals, and complex types such as Array/Struct/Map) are cached but do not contribute min/max bounds, so they only record null counts and sizes.
These statistics enable partition pruning when filtering:
val df = spark.range(10000000).cache() // This filter can skip batches using min/max statistics df.filter("id > 5000000").count()
The cached data itself lives on the JVM heap, not off-heap. Each cached batch is stored as a serialized Arrow IPC byte array (Array[Byte]), and the default Dataset.cache() storage level is the deserialized MEMORY_AND_DISK, so those bytes are retained as ordinary heap objects (and spill to disk under memory pressure). Arrow's off-heap allocators are used only for the transient VectorSchemaRoots created while encoding a batch for caching and while decoding a batch on read; these are released as soon as the encode/decode completes.
Sizing implications:
spark.executor.memoryOverhead covers the transient off-heap encode/decode buffers, which are proportional to a single batch, not to the total cached size. It generally does not need to grow with the size of the cache.Memory Cleanup:
You can monitor cache block sizes through the Storage tab in the Spark UI.
The cache serializer is resolved from spark.sql.cache.serializer only on first use and is then held in a process-wide field that is not reset when a SparkSession stops. As a result, switching cache formats requires a fresh JVM once any cache has been materialized -- stopping and rebuilding the SparkSession in the same process keeps using the originally resolved serializer.
To migrate from the default cache to Arrow cache:
val spark = SparkSession.builder() .config("spark.sql.cache.serializer", "org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer") .getOrCreate()
Note: Cache data is never shared across formats; each application caches in whichever format its serializer produces.
If you encounter OOM errors with Arrow cache:
Reduce batch size:
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "5000") // Default: 10000
Enable compression:
spark.conf.set("spark.sql.execution.arrow.compression.codec", "zstd")
Reduce compression level:
spark.conf.set("spark.sql.execution.arrow.compression.zstd.level", "1")
If Arrow cache is slower than expected:
Enable vectorized reader:
spark.conf.set("spark.sql.inMemoryColumnarStorage.enableVectorizedReader", "true")
Reduce or disable compression (decompression is part of every read):
spark.conf.set("spark.sql.execution.arrow.compression.zstd.level", "1") // faster, less ratio // or, for read-heavy workloads where memory is not the constraint: spark.conf.set("spark.sql.execution.arrow.compression.codec", "none")
Note: lz4 is not recommended. Arrow's Java LZ4 codec always uses the pure-Java Commons Compress framed LZ4 implementation, which is far slower than zstd.
Increase batch size (if memory allows):
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "20000")
| Configuration | Default | Description |
|---|---|---|
spark.sql.cache.serializer | DefaultCachedBatchSerializer | Cache format serializer class |
spark.sql.execution.arrow.compression.codec | none | Compression codec (none, lz4, zstd) |
spark.sql.execution.arrow.compression.zstd.level | 3 | Zstd compression level (negative = faster, up to 22) |
spark.sql.execution.arrow.maxRecordsPerBatch | 10000 | Maximum rows per Arrow batch |
spark.sql.execution.arrow.maxBytesPerBatch | 64MB | Maximum bytes per Arrow batch (whichever limit is hit first applies) |
spark.sql.execution.arrow.cache.prefetch.enabled | false | Prefetch the next batch in the background while the current one is consumed |
spark.sql.inMemoryColumnarStorage.enableVectorizedReader | true | Enable vectorized cache reading |
import org.apache.spark.sql.SparkSession object ArrowCacheExample { def main(args: Array[String]): Unit = { // Create SparkSession with Arrow cache val spark = SparkSession.builder() .appName("ArrowCacheExample") .master("local[*]") .config("spark.sql.cache.serializer", "org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer") .config("spark.sql.execution.arrow.compression.codec", "zstd") .config("spark.sql.inMemoryColumnarStorage.enableVectorizedReader", "true") .getOrCreate() try { // Read columnar data source val df = spark.read.parquet("large_dataset.parquet") // Cache with Arrow format df.cache() // Queries benefit from zero-copy reads and statistics val result1 = df.filter("age > 30").select("name", "age").count() println(s"Filtered count: $result1") val result2 = df.groupBy("country").agg(sum("sales")).collect() println(s"Aggregation result: ${result2.mkString(", ")}") // Uncache when done df.unpersist() } finally { spark.stop() } } }
none for read-heavy workloads, or zstd (default level 3) when memory matters; avoid lz4 (Arrow's Java LZ4 codec is pure-Java Commons Compress and much slower than zstd)