[
https://issues.apache.org/jira/browse/TEZ-4760?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denys Kuzmenko updated TEZ-4760:
--------------------------------
Description:
`IFile.Writer` writes every record as several small writes (the key and value
lengths as vints, then the key and value bytes) directly into the codec stream,
or into the checksum stream when there is no codec. Each call is a full trip
into the stream: with Snappy, a synchronized `Compressor.setInput` and friends
per call; without a codec, the checksum update per call.
The patch puts an unsynchronized 64 KB write buffer behind a
`NonSyncDataOutputStream` between the writer and those streams, so the
per-record writes land in one array and reach the codec in large chunks.
`close()` drains the buffer before it finishes the codec and checksum streams.
A Writer is confined to one thread, so the buffer takes no lock
(`BufferedOutputStream` locks on every call).
Microbenchmark (JDK 21, 6 runs each, median CPU per record; base = master):
| Path | Rows | Snappy | No codec |
| IFile.Writer | default | -12.5 % | -67.6 % |
| IFile.Writer | small | -33 % | -72 % |
| UnorderedPartitionedKVWriter, 24 partitions | default / small | -11.1 % /
-11.8 % | |
| UnorderedPartitionedKVWriter, 1,009 partitions | default / small | -5.8 % /
-12.6 % | |
Context: in Hive on Tez (LLAP), TPC-DS 10 TB, the shuffle write path
(ReduceSink serialization, then UnorderedPartitionedKVWriter, IFile and Snappy)
is 46-72 % of task CPU in the shuffle-heavy queries.
was:
`IFile.Writer` writes every record as several small writes (the key and value
lengths as vints, then the key and value bytes) directly into the codec stream,
or into the checksum stream when there is no codec. Each call is a full trip
into the stream: with Snappy, a synchronized `Compressor.setInput` and friends
per call; without a codec, the checksum update per call.
The patch puts an unsynchronized 64 KB write buffer behind a
`NonSyncDataOutputStream` between the writer and those streams, so the
per-record writes land in one array and reach the codec in large chunks.
`close()` drains the buffer before it finishes the codec and checksum streams.
A Writer is confined to one thread, so the buffer takes no lock
(`BufferedOutputStream` locks on every call).
Microbenchmark (JDK 21, 6 runs each, median CPU per record; base = master):
| Path | Rows | Snappy | No codec |
| IFile.Writer | default | -12.5 % | -67.6 % |
| IFile.Writer | small | -33 % | -72 % |
| UnorderedPartitionedKVWriter, 24 partitions | default / small | -11.1 % /
-11.8 % | |
| UnorderedPartitionedKVWriter, 1,009 partitions | default / small | -5.8 % /
-12.6 % | |
> IFile.Writer: buffer per-record writes before the compression and checksum
> streams
> ----------------------------------------------------------------------------------
>
> Key: TEZ-4760
> URL: https://issues.apache.org/jira/browse/TEZ-4760
> Project: Apache Tez
> Issue Type: Bug
> Reporter: Denys Kuzmenko
> Priority: Major
>
> `IFile.Writer` writes every record as several small writes (the key and value
> lengths as vints, then the key and value bytes) directly into the codec
> stream, or into the checksum stream when there is no codec. Each call is a
> full trip into the stream: with Snappy, a synchronized `Compressor.setInput`
> and friends per call; without a codec, the checksum update per call.
> The patch puts an unsynchronized 64 KB write buffer behind a
> `NonSyncDataOutputStream` between the writer and those streams, so the
> per-record writes land in one array and reach the codec in large chunks.
> `close()` drains the buffer before it finishes the codec and checksum
> streams. A Writer is confined to one thread, so the buffer takes no lock
> (`BufferedOutputStream` locks on every call).
> Microbenchmark (JDK 21, 6 runs each, median CPU per record; base = master):
> | Path | Rows | Snappy | No codec |
> | IFile.Writer | default | -12.5 % | -67.6 % |
> | IFile.Writer | small | -33 % | -72 % |
> | UnorderedPartitionedKVWriter, 24 partitions | default / small | -11.1 % /
> -11.8 % | |
> | UnorderedPartitionedKVWriter, 1,009 partitions | default / small | -5.8 % /
> -12.6 % | |
> Context: in Hive on Tez (LLAP), TPC-DS 10 TB, the shuffle write path
> (ReduceSink serialization, then UnorderedPartitionedKVWriter, IFile and
> Snappy) is 46-72 % of task CPU in the shuffle-heavy queries.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
