wombatu-kun commented on code in PR #19550:
URL: https://github.com/apache/hudi/pull/19550#discussion_r3922383489
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SparkSampleWritesUtils.java:
##########
@@ -109,12 +125,16 @@ private static Pair<Boolean, String>
doSampleWrites(JavaSparkContext jsc, Option
Pair<Boolean, String> emptyRes = Pair.of(false, null);
try (SparkRDDWriteClient sampleWriteClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), sampleWriteConfig, Option.empty())) {
int size = writeConfig.getIntOrDefault(SAMPLE_WRITES_SIZE);
+ long maxSampleBytes = resolveMaxSampleBytes(jsc);
return recordsOpt.map(records -> {
- // Empty partition path so all sampled records write to a single
non-partitioned file,
- // instead of fanning out into one tiny file per source partition and
skewing the estimate.
- List<HoodieRecord> samples = records.coalesce(1).take(size).stream()
- .map(r -> r.newInstance(new HoodieKey(r.getRecordKey(), "")))
- .collect(Collectors.toList());
+ // Collapse to a single partition, then take a sample bounded by both
record count and total
+ // serialized bytes on the executor. The sample is later shipped
inside one Spark task, so
+ // bounding its serialized size keeps that task under
spark.rpc.message.maxSize even when the
+ // source records are large.
+ List<HoodieRecord> samples = records.coalesce(1)
Review Comment:
This already yields a one-partition RDD of re-keyed samples, and it is
collected to the driver only to be re-parallelized into `bulkInsert` - which is
the step the RPC frame limit actually binds on. Would persisting that RDD and
passing it straight to `bulkInsert` work instead, removing the limit from the
picture rather than budgeting against it?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SparkSampleWritesUtils.java:
##########
@@ -109,12 +125,16 @@ private static Pair<Boolean, String>
doSampleWrites(JavaSparkContext jsc, Option
Pair<Boolean, String> emptyRes = Pair.of(false, null);
try (SparkRDDWriteClient sampleWriteClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), sampleWriteConfig, Option.empty())) {
int size = writeConfig.getIntOrDefault(SAMPLE_WRITES_SIZE);
+ long maxSampleBytes = resolveMaxSampleBytes(jsc);
return recordsOpt.map(records -> {
- // Empty partition path so all sampled records write to a single
non-partitioned file,
- // instead of fanning out into one tiny file per source partition and
skewing the estimate.
- List<HoodieRecord> samples = records.coalesce(1).take(size).stream()
- .map(r -> r.newInstance(new HoodieKey(r.getRecordKey(), "")))
- .collect(Collectors.toList());
+ // Collapse to a single partition, then take a sample bounded by both
record count and total
+ // serialized bytes on the executor. The sample is later shipped
inside one Spark task, so
+ // bounding its serialized size keeps that task under
spark.rpc.message.maxSize even when the
+ // source records are large.
+ List<HoodieRecord> samples = records.coalesce(1)
+ .mapPartitions((FlatMapFunction<Iterator<HoodieRecord>,
HoodieRecord>) sourceRecords ->
+ takeBoundedSample(sourceRecords, size, maxSampleBytes))
Review Comment:
`hoodie.streamer.sample.writes.enabled` defaults to `false` and
`hoodie.streamer.sample.writes.size` to 5000, not `true` and 25000, so the
feature is opt-in and the per-record size at which the sample task outgrows the
RPC limit is around 27KB rather than the ~5.4KB derived in the description and
in #19549. Worth correcting in the commit message too, since that is what lands
on master.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SparkSampleWritesUtils.java:
##########
@@ -139,6 +159,46 @@ private static Pair<Boolean, String>
doSampleWrites(JavaSparkContext jsc, Option
}
}
+ /**
+ * Resolves the maximum total serialized size (in bytes) allowed for the
sampled records, derived
+ * as {@link #SAMPLE_WRITES_TASK_BYTES_FRACTION} of the cluster's {@code
spark.rpc.message.maxSize}
+ * (read from {@link
org.apache.spark.internal.config.Network#RPC_MESSAGE_MAX_SIZE}, which also
+ * supplies Spark's default). Deriving it from the RPC limit keeps the bound
correct if operators
+ * raise that limit.
+ */
+ private static long resolveMaxSampleBytes(JavaSparkContext jsc) {
+ int rpcMaxSizeMb = (Integer)
jsc.getConf().get(Network$.MODULE$.RPC_MESSAGE_MAX_SIZE());
Review Comment:
`org.apache.spark.util.RpcUtils.maxMessageSizeBytes(SparkConf)` is a plain
public static across Spark 3.3 to 4.1 that returns this same value in bytes,
and it carries Spark's own guard against a limit above 2047MB. Calling it would
drop the `Network$` import, the `(Integer)` cast and `BYTES_PER_MB` in one go.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SparkSampleWritesUtils.java:
##########
@@ -109,12 +125,16 @@ private static Pair<Boolean, String>
doSampleWrites(JavaSparkContext jsc, Option
Pair<Boolean, String> emptyRes = Pair.of(false, null);
try (SparkRDDWriteClient sampleWriteClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), sampleWriteConfig, Option.empty())) {
int size = writeConfig.getIntOrDefault(SAMPLE_WRITES_SIZE);
+ long maxSampleBytes = resolveMaxSampleBytes(jsc);
return recordsOpt.map(records -> {
- // Empty partition path so all sampled records write to a single
non-partitioned file,
- // instead of fanning out into one tiny file per source partition and
skewing the estimate.
- List<HoodieRecord> samples = records.coalesce(1).take(size).stream()
- .map(r -> r.newInstance(new HoodieKey(r.getRecordKey(), "")))
- .collect(Collectors.toList());
+ // Collapse to a single partition, then take a sample bounded by both
record count and total
+ // serialized bytes on the executor. The sample is later shipped
inside one Spark task, so
+ // bounding its serialized size keeps that task under
spark.rpc.message.maxSize even when the
+ // source records are large.
+ List<HoodieRecord> samples = records.coalesce(1)
+ .mapPartitions((FlatMapFunction<Iterator<HoodieRecord>,
HoodieRecord>) sourceRecords ->
+ takeBoundedSample(sourceRecords, size, maxSampleBytes))
+ .collect();
Review Comment:
`getWriteConfigWithRecordSizeEstimate` catches only `IOException` and
`syncOnce` has no catch at all, so the RPC-size failure aborts the streamer
round rather than falling back to the default estimate as the description says.
Worth restating the impact, and worth deciding whether that catch should widen
so a size estimate can never fail ingestion.
--
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]
