Jakub Wozniak created SPARK-58060:
-------------------------------------
Summary: Reading array columns from Parquet files with large data
pages extremely slow
Key: SPARK-58060
URL: https://issues.apache.org/jira/browse/SPARK-58060
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 3.5.3
Environment: Spark 3.5.3, Java 11.
Reporter: Jakub Wozniak
## Summary
Reading a Parquet file whose data pages are large (which `parquet-mr` produces
for tables
with few rows but very large `array`/`list` values) is **~100–270× slower**
than reading the
exact same data written with small pages, and than reading it with pyarrow. The
cost is in
the vectorized Parquet page-decode path and no reader configuration mitigates
it.
## Steps to reproduce (self-contained, no external data)
{code:python}
import time
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
N_ROWS, N_ELEM = 24, 1_000_000 # 24 rows x 1M doubles/row (~192MB)
spark = (SparkSession.builder.master('local[4]').appName('pagesize-repro')
.config('spark.driver.memory', '16g')
.config('spark.driver.extraJavaOptions', '-XX:MaxDirectMemorySize=8g')
.getOrCreate())
df = (spark.range(N_ROWS)
.select(F.transform(F.sequence(F.lit(1), F.lit(N_ELEM)),
lambda i: F.rand()).alias('wave')))
def write_read(path, opts, label, n_reads=3):
hc = spark._jsc.hadoopConfiguration()
hc.set('parquet.page.size.row.check.min', '100') # parquet-mr defaults
hc.set('parquet.page.size.row.check.max', '10000')
hc.set('parquet.page.size', str(1024*1024))
for k, v in opts.items():
hc.set(k, v)
df.coalesce(1).write.mode('overwrite').parquet('file://'+path)
for i in range(n_reads):
s = time.time()
spark.read.parquet('file://'+path).select('wave').write.format('noop').mode('overwrite').save()
print(f"[{time.time()-s:8.1f}s] read #{i+1} {label}", flush=True)
# Read the small-page file FIRST: its 3 reads are fast and fully warm the JVM
(JIT,
# whole-stage codegen, the general Parquet read path). The big-page file's
FIRST read is
# still ~150x slower despite that warm-up -> the penalty is specific to the
giant-page
# decode path and is not relieved by warming on normal (small-page) data.
write_read('/tmp/repro_smallpage', {'parquet.page.size.row.check.min':'1',
'parquet.page.size.row.check.max':'1'},
"small-page file (many small pages) -- warms the JVM")
write_read('/tmp/repro_bigpage', {},
"BIG-page file (parquet-mr default, few huge pages) -- read #1 STILL
slow after warmup")
spark.stop()
{code}
## Actual result (Spark 3.5.3)
{noformat}
[ 1.2s] read #1 small-page file (many small pages) -- warms the JVM
[ 0.7s] read #2 small-page file (many small pages) -- warms the JVM
[ 0.5s] read #3 small-page file (many small pages) -- warms the JVM
[ 142.4s] read #1 BIG-page file (parquet-mr default) -- read #1 STILL slow
after warmup
[ 1.9s] read #2 BIG-page file (parquet-mr default)
[ 1.2s] read #3 BIG-page file (parquet-mr default)
{noformat}
Same data, same reader. Three fast small-page reads fully warm the JVM, yet the
big-page
file's first read is still ~150× slower — then its own subsequent reads are
fast. This
shows the cost is (a) specific to large data pages and (b) a
giant-page-specific cold JIT
cost, not relieved by warming on normal data.
## Expected result
The first read of a Parquet file with large data pages should be comparable to
reading the
same data with small pages (and to pyarrow, which reads the equivalent 1 GB
real-world file
in ~4 s where Spark takes 5–18 min).
## Why this matters (real-world trigger)
`parquet-mr` decides when to flush a data page using a **row-count** check
(`parquet.page.size.row.check.min`, default 100 rows), estimating page bytes
only every N
rows. A table of wide rows (e.g. one `array<float>` of ~2M elements ≈ 8–15 MB
per row, a few
dozen rows per row group — common for scientific/waveform data) therefore gets
**a few
hundred-MB data pages**. Any Spark job reading such files — produced by Spark's
own writer or
any `parquet-mr` producer — hits this. pyarrow flushes pages by bytes and is
unaffected.
## Isolation performed
1. **Real 1 GB file** (170 rows, `array<float>` ~7.6 MB/row): Spark 333 s vs
pyarrow 4.4 s;
an 85-row sibling: Spark 1065 s vs pyarrow 3.9 s. Scalar columns from the
same file: 0.1 s.
2. **Writer is the only variable:** same 30×2M-float data written by pyarrow
reads in Spark in
1.6 s; written by `parquet-mr` (Spark's writer) reads in 218 s.
3. **`parquet.page.size` sweep** (Spark writes, Spark reads): 1 MB → 1.1 s, 8
MB → 1.1 s,
64 MB → 14.4 s, 256 MB → 52 s. Superlinear in page size.
4. **No reader config mitigates** (real 1 GB file, all ~330–410 s):
`spark.sql.parquet.enableVectorizedReader=false`,
`enableNestedColumnVectorizedReader=false`, `columnarReaderBatchSize=1` and
`=16`.
Both the vectorized and the row-based readers are affected.
5. **Mitigation confirmed on the real data:** rewriting the file with
byte-flushed 1 MB pages
(`parquet.page.size.row.check.min=1`) turns the 333 s read into 2.9 s.
## Observations that may help diagnosis
These are just what we measured while narrowing it down; we may well be
misreading them, and
we defer to the maintainers on the actual cause. The slow read appears to be a
*cold
first-read* cost within a JVM rather than steady-state decode cost:
- Re-reading the same file in 5 consecutive *fresh* JVMs is slow every time (so
it does not
look like OS page cache).
- The 142 s cold read shows 0 GC collections / 0 ms GC time and a flat heap,
i.e. it seems
CPU-bound rather than allocation/GC-bound.
- Within one JVM, after the first big-page read, a *different* big-page file
reads in ~1 s —
suggesting the warm-up is code-level rather than data/file-specific.
- Reading small-page files first does **not** speed up the first big-page read
(still ~140 s),
so whatever warms up seems specific to the large-page path.
Cold-read stack samples (single worker thread) were consistently in the page
decoder:
{noformat}
VectorizedColumnReader.readBatch -> readPage -> readPageV1 ->
VectorizedColumnReader$1.visit
java.io.DataInputStream.readFully
jdk.internal.misc.Unsafe.setMemory
{noformat}
One *possible* reading (very much a guess) is that a single large page is
decoded in one long
`readPageV1` invocation, and the intrinsic-backed operations there
(`Unsafe.setMemory`,
`DataInputStream.readFully`) are cheap once JIT-compiled but expensive on the
first,
interpreted pass — with many small pages the path stays warm from normal
workloads. We are
not confident in this and would appreciate the maintainers' assessment.
## Potential directions (for maintainers to weigh)
We are not sure what the right fix is; a few directions that *might* be worth
considering:
- Whether the large-page decode path could be exercised/compiled earlier, or
processed in
bounded chunks, so the first large page is not decoded on a cold path.
- Whether this is considered a reader issue at all, or expected given how
`parquet-mr` sizes
pages (its default row-count-based page flush produces very large pages for
wide-row tables).
- At minimum, documenting the interaction may help others who hit it.
## Workaround (works for us today)
Writing these tables with byte-based page flushing so pages stay ~1 MB avoids
the slow read
entirely (on our real data this changed a 333 s read to 2.9 s):
{noformat}
parquet.page.size.row.check.min = 1
parquet.page.size.row.check.max = 1
parquet.page.size = 1048576
{noformat}
Performance of writes with such settings yet to be checked...
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
