[
https://issues.apache.org/jira/browse/SPARK-58060?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jakub Wozniak updated SPARK-58060:
----------------------------------
Description:
*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 (to be checked)*
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*
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...
was:
## 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...
> 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
> Priority: Major
> Labels: parquet, perfomance, vectorization
>
> *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 (to be checked)*
> 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*
> 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]
