loustler opened a new pull request, #11661:
URL: https://github.com/apache/seatunnel/pull/11661
## What changed
`AbstractWriteStrategy#getConfiguration` builds a Hadoop `Configuration`
from scratch on every
call. It is called **once per output file**, so a subtask that rotates files
re-does the work for
every file it writes.
The expensive part is not the object — it is the resource load. The first
property access on a new
`Configuration` parses `core-default.xml` and the other default resources.
`HadoopConf#toConfiguration`
touches a property immediately (`setBoolean`), so every call pays the parse.
This caches the parsed instance per write strategy and hands out copies.
Hadoop's copy constructor
clones the already-loaded `Properties` instead of re-reading the XML, so
callers still get a fresh,
independently mutable `Configuration` — which matters, because some of them
mutate it
(`ParquetWriteStrategy#init` sets `AvroWriteSupport.WRITE_FIXED_AS_INT96` on
the instance it
receives).
```java
@Override
public Configuration getConfiguration(HadoopConf hadoopConf) {
+ // The cache is only valid for the HadoopConf this strategy was
initialised with.
+ if (hadoopConf != this.hadoopConf) {
+ return buildConfiguration(hadoopConf);
+ }
+ if (parsedConfiguration == null) {
+ parsedConfiguration = buildConfiguration(hadoopConf);
+ }
+ return new Configuration(parsedConfiguration);
+}
+
+private Configuration buildConfiguration(HadoopConf hadoopConf) {
Configuration configuration = hadoopConf.toConfiguration();
this.hadoopConf.setExtraOptionsForConfiguration(configuration);
return configuration;
}
```
Affected call sites, all per output file or per init — `getConfiguration` is
never on a per-row path:
| call site | frequency |
|---|---|
| `ParquetWriteStrategy:190` (`HadoopOutputFile.fromPath`) | per output file
|
| `OrcWriteStrategy:149` (`OrcFile.writerOptions`) | per output file |
| `ParquetWriteStrategy:101` (`init`, feeds `AvroSchemaConverter`) | once
per subtask |
## Measured
`ConfigBench` (source in the collapsed block at the end) against the
classpath a Zeta worker actually loads — the shipped
`lib/seatunnel-hadoop3-*-uber.jar`, which bundles `core-default.xml` (152
KB, 462 properties) plus
every transitive dependency. JDK 8 `1.8.0_502`, `-Xms2g -Xmx2g`, 3 s warmup,
7 × 2 s rounds,
medians. `TOCONF` is the current per-file path; `COPY` is what this PR does.
| mode | 1 thread ops/s | 8 threads ops/s | CPU ms/op | allocated B/op |
|---|---:|---:|---:|---:|
| `new Configuration()` alone | 1,432,625 | 1,481,705 | 0.0010 | 288 |
| **`toConfiguration()` — before** | 630 | 3,986 | **1.652** | **1,591,454**
|
| **`new Configuration(cached)` — after** | 71,699 | 68,931 | **0.033** |
**39,680** |
| reuse one instance (floor, not proposed) | 19,575,608 | — | 0.0001 | 24 |
So per output file: **1.65 ms → 0.033 ms of CPU** and **1,591,454 B → 39,680
B (−97.5 %)**. At one
thread this work is CPU-bound, so wall tracks CPU closely (1/630 s = 1.59
ms/op).
Two things that table shows which are worth stating plainly:
- **The parse is lazy.** `new Configuration()` on its own is 0.4 µs and
allocates 288 B. All of the
cost is the first property touch. That is why the fix is "parse once", not
"allocate less".
- **`Configuration` construction does not scale across threads.** The two
constructor-bound modes
come out at 0.96× and 1.03× on 8 threads. Both constructors take a monitor
on `Configuration.class`
to register into a static `WeakHashMap` (`monitorenter` at offset 97 in
`Configuration(boolean)`,
238 in the copy constructor), which caps construction at ~69k/s per JVM.
At per-file frequency
that is irrelevant, but the copy is not free and I would rather say so
than imply otherwise.
On the real write path (`ParquetWriteStrategy#write`, 9-column schema, 200k
rows, 1 thread, JDK 8,
7 rounds), patched/unpatched alternated across separate JVMs, **two
independent A/B pairs**:
`batch_size=2000` → 100 output files:
| metric | before | after | pair 1 | pair 2 |
|---|---:|---:|---:|---:|
| allocation / row | 4,817.7 / 4,850.9 B | 3,814.9 / 3,782.9 B | **−20.8 %**
| **−22.0 %** |
| CPU / row | 5,973.2 / 5,931.6 ns | 3,929.4 / 3,917.4 ns | **−34.2 %** |
**−34.0 %** |
| wall / row | 25,925 / 26,211 ns | 24,004 / 24,164 ns | **−7.4 %** | **−7.8
%** |
| throughput | 38,573 / 38,152 rows/s | 41,659 / 41,383 rows/s | **+8.0 %**
| **+8.5 %** |
No rotation → 1 output file (the control):
| metric | before | after | delta |
|---|---:|---:|---:|
| allocation / row | 2,726.8 B | 2,716.4 B | **−10.4 B** |
The control is what establishes that this is a per-file cost and not a
per-row one: with a single
output file the saving collapses from ~1,000 B/row to ~10 B/row. (In pair 2
the *unpatched*
no-rotation baseline drifted to 2,918.7 B/row while the patched value came
out at 2,716.4 B/row in
both pairs, identical to 0.1 B. I am quoting pair 1 for the control because
pair 2's baseline is the
outlier, not because it is the friendlier number.)
**Every metric came out larger in situ than the microbenchmark predicts, and
I have not isolated
why.** Comparing like with like — isolated saving is the before/after delta
of the two bold
`ConfigBench` rows, in-situ saving is the per-row delta × 2,000 rows per
file:
| saved per output file | isolated (ConfigBench) | in situ, pair 1 / pair 2
| ratio |
|---|---:|---:|---:|
| allocation | 1.552 MB | 2.006 / 2.136 MB | 1.29× / 1.38× |
| CPU | 1.619 ms | 4.088 / 4.028 ms | 2.5× / 2.5× |
| wall | 1.573 ms | 3.841 / 4.093 ms | 2.4× / 2.6× |
CPU and wall agree with each other tightly (~2.5×) and disagree with
allocation (~1.3×), and neither
ratio is an integer — so this is not `getConfiguration` being called more
times per output file than
I think it is. Something about handing `HadoopOutputFile`/`ParquetWriter` a
freshly-parsed
`Configuration` rather than a copy costs more downstream than the
construction itself. I did not
isolate it, so it is an observation, not a claim, and every figure quoted
elsewhere in this PR is the
smaller isolated one.
## What this does *not* claim
**The +8 % is a sink-loop number, not a job number.** It is measured with
the write loop as the only
work in the process. A real pipeline also reads, deserialises, and often
blocks on the source: in the
production run that started this investigation the writers were parked 56.7
% of the time, and no
sink-side saving can show up through that. So: +8 % on the write path, and
no end-to-end claim.
**`Configuration` is a small share of per-file overhead, not a dominant
one.** At `batch_size=2000`
the harness costs 25,925 ns/row — **51.9 ms of wall time per output file** —
against 2,589 ns/row
with no rotation. Of that 51.9 ms this recovers 1.57 ms by the isolated
measurement (3.0 %) or
3.8–4.1 ms as measured in situ (7.4–7.9 %). The rest is `ParquetWriter`
construction, opening the
file, and writing the footer on close, and this PR does not touch any of it.
An earlier internal
estimate put the `Configuration` cost at "~48 ms per
file"; that figure was back-computed from exactly this throughput gap and
silently attributed the
whole gap to `Configuration`. Direct measurement shows it was wrong by
roughly 30×. Correcting it
here rather than letting a reviewer find it.
Jobs writing many small files (small `batch_size`, high partition
cardinality) get proportionally
more of this; a job writing one large file per subtask gets ~10 B/row, which
is nothing.
Also not measured: any effect on HDFS/S3/OSS specifically — the harness
writes to a local
filesystem, and JDK 8 on this host is x86_64 under Rosetta, so the ratios
are meaningful and the
absolute numbers are not.
## Why copies and not one shared instance
Handing every caller the same cached instance would be faster still — 0.033
ms → 0.0001 ms. It was
not done, because `getConfiguration` currently guarantees a fresh instance
and at least one caller
relies on that: `ParquetWriteStrategy#init` mutates the returned
`Configuration`. Sharing it would
leak that mutation into the per-file configuration. Weighed against ~48 ms
of other per-file work,
saving a further 33 µs is not worth introducing shared mutable state into a
class that is not
thread-safe today.
The `hadoopConf != this.hadoopConf` guard exists because
`getConfiguration(HadoopConf)` is a public
interface method that takes its conf as a parameter. Every current caller
passes the field, so the
guard is defensive — but the alternative is a method that silently ignores
its own argument.
## Behaviour
- The returned object is still a distinct, freely mutable `Configuration` on
every call.
- It still carries everything `toConfiguration()` and
`setExtraOptionsForConfiguration()` set, plus
the parsed default resources.
- The field is `transient`: `WriteStrategy extends Serializable` and
`Configuration` is not.
Added `AbstractWriteStrategyConfigurationTest` covering exactly those three
points — independent
instances, mutation isolation across calls, and that a *different*
`HadoopConf` is not served from
the cache.
## Verification
- `mvn test -pl seatunnel-connectors-v2/connector-file/connector-file-base`
on JDK 8 `1.8.0_502`:
**240 tests, 0 failures, 0 errors** (237 before this PR + the 3 added
here). `spotless:check` runs
in that build and passes.
- Grepped for other overrides and callers of `getConfiguration`: nothing
outside
`connector-file-base` overrides `AbstractWriteStrategy#getConfiguration`,
and the three call sites
in the table above are all of them.
## Reproducing the benchmark
```bash
# uber jar out of any built dist; slf4j-api is needed because the uber jar
deliberately
# excludes slf4j (without it: NoClassDefFoundError org/slf4j/LoggerFactory)
CP=lib/seatunnel-hadoop3-3.4.3-uber.jar:starter/logging/slf4j-api-1.7.36.jar
javac -cp "$CP" -d out ConfigBench.java
for m in NEW_ONLY NEW_GET TOCONF COPY REUSE; do
java -Xms2g -Xmx2g -cp "$CP:out" -Dcb.mode=$m -Dcb.threads=1 ConfigBench |
grep CB_RESULT
done
```
The write-path numbers come from a throwaway JUnit harness that drives
`ParquetWriteStrategy#write`
directly and reports `getThreadAllocatedBytes` / `getCurrentThreadCpuTime`
per row. It is not part of
this PR — it is a measurement tool, not a test, and it would only rot in the
tree. Happy to attach it
if a reviewer wants to re-run the A/B.
<details>
<summary><code>ConfigBench.java</code> — the full benchmark source</summary>
```java
import com.sun.management.ThreadMXBean;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.hadoop.conf.Configuration;
/**
* Measure what a per-output-file Hadoop {@link Configuration} construction
actually costs, so
* the SeaTunnel file-sink fix can be argued from a number rather than from
a derived estimate.
*
* Run against the SHIPPED uber jar (lib/seatunnel-hadoop3-*-uber.jar out of
the dist tarball),
* not against a hand-assembled classpath: the uber jar is what production
loads, it bundles
* core-default.xml and every transitive dependency, and its relocations are
part of the thing
* being measured.
*
* javac -cp uber.jar -d out ConfigBench.java
* java -cp uber.jar:out -Dcb.mode=TOCONF -Dcb.threads=1 ConfigBench
*
* Modes:
* NEW_ONLY new Configuration() - shows the XML
parse is lazy
* NEW_GET new Configuration() + get() - forces the parse
* TOCONF replica of HadoopConf#toConfiguration() - what SeaTunnel
does per output file
* COPY new Configuration(warmSource) + get() - the proposed fix's
residual cost
* REUSE hand back one prebuilt instance - the zero-cost floor
*
* Why both 1 and 8 threads: BOTH constructors take a monitor on
Configuration.class to register
* into a static WeakHashMap (verified in hadoop-common 3.4.3 bytecode:
monitorenter at offset 97
* in Configuration(boolean), 238 in the copy constructor). Per output file
that should be
* irrelevant; the 8-thread column is there to confirm it, not to assume it.
*/
public class ConfigBench {
/** Written to on every op so the JIT cannot fold the work away. */
static volatile Object blackhole;
enum Mode {
NEW_ONLY,
NEW_GET,
TOCONF,
COPY,
REUSE
}
// The six properties HadoopConf#toConfiguration() sets, with the keys
SeaTunnel uses for
// plain HDFS. Values are irrelevant to the cost; the count and the
first-touch are not.
static Configuration toConfiguration() {
Configuration c = new Configuration();
c.setBoolean("parquet.avro.readInt96AsFixed", true);
c.setBoolean("parquet.avro.add-list-element-records", false);
c.setBoolean("parquet.avro.write-old-list-structure", true);
c.setBoolean("fs.hdfs.impl.disable.cache", true);
c.set("fs.defaultFS", "hdfs://localhost:9000");
c.set("fs.hdfs.impl",
"org.apache.hadoop.hdfs.DistributedFileSystem");
return c;
}
static Object op(Mode mode, Configuration warm, Configuration reuse) {
switch (mode) {
case NEW_ONLY:
return new Configuration();
case NEW_GET: {
Configuration c = new Configuration();
return c.get("fs.defaultFS");
}
case TOCONF:
return toConfiguration();
case COPY: {
Configuration c = new Configuration(warm);
return c.get("fs.defaultFS");
}
case REUSE:
return reuse.get("fs.defaultFS");
default:
throw new IllegalStateException();
}
}
public static void main(String[] args) throws Exception {
Mode mode = Mode.valueOf(System.getProperty("cb.mode", "TOCONF"));
int threads = Integer.getInteger("cb.threads", 1);
int rounds = Integer.getInteger("cb.rounds", 7);
long roundMs = Long.getLong("cb.roundms", 2000L);
long warmupMs = Long.getLong("cb.warmupms", 3000L);
String label = System.getProperty("cb.label", mode + "-t" + threads);
// A source instance whose resources are already parsed, i.e. what a
cached
// Configuration in AbstractWriteStrategy would look like at steady
state.
final Configuration warm = toConfiguration();
warm.get("fs.defaultFS");
final Configuration reuse = toConfiguration();
reuse.get("fs.defaultFS");
System.out.printf(
"CB_START label=%s mode=%s threads=%d rounds=%d roundMs=%d
jdk=%s%n",
label, mode, threads, rounds, roundMs,
System.getProperty("java.version"));
// ---- warmup ----
long wEnd = System.nanoTime() + warmupMs * 1_000_000L;
long wOps = 0;
while (System.nanoTime() < wEnd) {
blackhole = op(mode, warm, reuse);
wOps++;
}
System.out.printf("CB_WARMUP label=%s ops=%d%n", label, wOps);
ThreadMXBean tmx = (ThreadMXBean)
ManagementFactory.getThreadMXBean();
List<Double> opsPerSec = new ArrayList<>();
List<Double> nsPerOp = new ArrayList<>();
List<Double> bytesPerOp = new ArrayList<>();
for (int r = 0; r < rounds; r++) {
final CountDownLatch go = new CountDownLatch(1);
final long[] ops = new long[threads];
final long[] alloc = new long[threads];
final long[] cpu = new long[threads];
Thread[] ts = new Thread[threads];
for (int i = 0; i < threads; i++) {
final int idx = i;
ts[i] =
new Thread(
() -> {
try {
go.await();
} catch (InterruptedException e) {
return;
}
long id = Thread.currentThread().getId();
long a0 =
tmx.getThreadAllocatedBytes(id);
long c0 = tmx.getCurrentThreadCpuTime();
long end = System.nanoTime() + roundMs *
1_000_000L;
long n = 0;
Object local = null;
while (System.nanoTime() < end) {
local = op(mode, warm, reuse);
n++;
}
blackhole = local;
ops[idx] = n;
alloc[idx] =
tmx.getThreadAllocatedBytes(id) - a0;
cpu[idx] = tmx.getCurrentThreadCpuTime()
- c0;
},
"cb-" + i);
ts[i].start();
}
long t0 = System.nanoTime();
go.countDown();
for (Thread t : ts) {
t.join();
}
long wall = System.nanoTime() - t0;
long totalOps = 0, totalAlloc = 0, totalCpu = 0;
for (int i = 0; i < threads; i++) {
totalOps += ops[i];
totalAlloc += alloc[i];
totalCpu += cpu[i];
}
double ops_s = totalOps / (wall / 1e9);
double ns_op = (double) totalCpu / totalOps;
double b_op = (double) totalAlloc / totalOps;
opsPerSec.add(ops_s);
nsPerOp.add(ns_op);
bytesPerOp.add(b_op);
System.out.printf(
"CB_ROUND label=%s r=%d ops=%d ops_s=%.1f cpu_ns_op=%.1f
alloc_b_op=%.1f%n",
label, r, totalOps, ops_s, ns_op, b_op);
}
System.out.printf(
"CB_RESULT label=%s mode=%s threads=%d ops_s_median=%.1f
cpu_ns_op_median=%.1f "
+ "alloc_b_op_median=%.1f ms_per_op_median=%.4f%n",
label,
mode,
threads,
median(opsPerSec),
median(nsPerOp),
median(bytesPerOp),
median(nsPerOp) / 1e6);
}
static double median(List<Double> xs) {
double[] a = new double[xs.size()];
for (int i = 0; i < a.length; i++) {
a[i] = xs.get(i);
}
Arrays.sort(a);
int n = a.length;
return n % 2 == 1 ? a[n / 2] : (a[n / 2 - 1] + a[n / 2]) / 2.0;
}
}
```
</details>
--
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]
