andygrove opened a new issue, #6944:
URL: https://github.com/apache/datafusion-comet/issues/6944
### What is the problem the feature request solves?
#6940 adds S3A magic committer support to native Parquet writes by letting
S3A create each file. The JVM opens the file with Hadoop's `FileSystem.create`,
and the native writer streams its encoded Parquet bytes into that stream over
JNI. S3A then turns the file into the pending multipart upload that the
committer completes at job commit, and applies its own write settings. The
encoding is native. The last leg of the bytes, and the upload itself, are not.
The JNI leg costs little. I measured the native writer writing local files
directly against the same writer writing them through the #6940 route. The
route writes to local disk through Hadoop's `RawLocalFileSystem` under its own
scheme, so only the route differs. Medians of 5 interleaved rounds:
| Codec | Output | Native, direct | Native, through the JVM | Spark's own
writer |
| ------------ | ------- | ------------------------ |
------------------------ | ------------------------ |
| snappy | 269 MiB | 2176 ms wall, 2410 ms CPU | 2172 ms wall, 2407 ms CPU |
5830 ms wall, 6537 ms CPU |
| uncompressed | 643 MiB | 1896 ms wall, 1964 ms CPU | 1926 ms wall, 1982 ms
CPU | 5813 ms wall, 6538 ms CPU |
The difference is within run-to-run noise, about 1% of CPU uncompressed,
which is the route's worst case. GC time does not change. Per file, the route
costs:
- two extra copies of the encoded bytes, into the writer's 1 MiB buffer and
then into a Java array;
- one JNI upcall per MiB;
- 1 MiB of native memory, counted against the task's memory pool, and one 1
MiB Java array, allocated once per file.
Setup: macOS arm64, Spark 4.1.3, `libcomet` built `--release` without LTO
and with 16 codegen units, `local[2]`, 20M rows of five columns (long, int,
double, string, long) read through a native scan, two write tasks, and a
warm-up round.
What the route leaves in the JVM is the upload, and it is exactly the upload
Spark's writer makes. By default S3A stages each file on local disk in 64 MiB
blocks, up to four per file (`fs.s3a.fast.upload.buffer=disk`,
`fs.s3a.multipart.size`, `fs.s3a.fast.upload.active.blocks`), and uploads them
with the AWS Java SDK. A native upload through `object_store` would skip the
local disk and the JVM entirely. It would hold parts in native memory instead:
the uploader in #6355 keeps up to nine 10 MiB parts per file. Nobody has
measured whether that is faster or cheaper on real S3.
### Describe the potential solution
Options, from the least to the most of S3A's job taken over:
1. **Keep the JVM route and trim it.** Writing straight into the Java array
would save one of the two extra copies. Given the numbers above, this alone is
worth little.
2. **Upload natively, and record the commit through S3A.** The native writer
uploads the parts and, on close, hands the upload ID and part ETags to the JVM.
The JVM records the pending commit through the tracker S3A would have created
for the path, so S3A still writes its own `.pending` file, or its in-memory
entry under `fs.s3a.committer.magic.track.commits.in.memory.enabled`. This
leaves S3A's commit formats with S3A, but it reaches into non-public S3A
classes (`MagicCommitIntegration`, `PutTracker`). The upload settings would
also have to be reproduced natively, since S3A applies them when it starts an
upload. A variant has the JVM start the upload through S3A and pass the upload
ID to native code, which then only uploads parts.
3. **Reproduce the magic committer's file protocol natively.** For a file
under a magic path, the native writer would compute the final key as
`MagicCommitPaths.finalDestination` does, start a multipart upload to it, and
upload the parts. On close it would write what
`S3MagicCommitTracker.aboutToComplete` writes: a zero-byte marker at the magic
path with the file's length in the `x-hadoop-s3a-magic-data-length` header, and
the `.pending` JSON (`SinglePendingCommit`: upload ID, bucket, destination key,
part ETags, length). Task and job commit stay with S3A. `object_store`'s
`MultipartStore` exposes the upload ID and ETags this needs, while
`put_multipart` hides them. The costs:
- Comet would depend on S3A's internal, versioned `.pending` format and
on path rules that differ across Hadoop lines (`__magic` on 3.3,
`__magic_job-<id>` on 3.4).
- It cannot serve the in-memory tracker, where task commit reads S3A's
in-process state rather than `.pending` files, so that setting would keep the
JVM route.
- Every S3A setting that applies to an upload would have to be translated
to `object_store`, or force the JVM route: encryption, canned ACL, storage
class, content encoding, `fs.s3a.create.header.*`, checksums. `object_store`
0.13 supports SSE-KMS, DSSE-KMS, SSE-C, storage class, content encoding and
user metadata, but not canned ACLs.
4. **A Comet commit protocol.** A `FileCommitProtocol` of Comet's own for S3
destinations: tasks upload natively and return their upload IDs and ETags in
their commit messages, and job commit completes the uploads. The bytes are
native end to end and nothing depends on S3A internals. But Comet would own a
committer, including speculation, task retries, cleanup after a job abort, and
`_SUCCESS` compatibility, and users would have to configure it in place of
S3A's magic committer.
The suggested next step is a benchmark of S3A's upload against
`object_store`'s on real S3: throughput, CPU, local disk I/O and memory, with
S3A's default `disk` buffering and with `bytebuffer`. #6929 asks for the same
measurement before sending every S3A write through S3A, and it decides whether
any option above is worth its coupling. If a native upload wins, option 3
restricted to the configurations it can reproduce, with the #6940 route as the
fallback, looks like the smallest step. It would share #6355's upload code,
moved from `put_multipart` to `MultipartStore`.
### Additional context
Follows #6940 and #6929. Related to #6355 and #1625.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
