lokeshj1703 commented on code in PR #19550:
URL: https://github.com/apache/hudi/pull/19550#discussion_r4036097499
##########
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:
Done. Reworked to hand the persisted single-partition RDD straight to
`bulkInsert`, so the sample never crosses the driver. `persist()` because it is
read more than once (isEmpty, bulkInsert, commit).
##########
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:
Fixed. The rewritten description no longer states those defaults, and the
change no longer hinges on the per-record threshold.
##########
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:
Addressed. The rework removes the driver round trip that caused the failure,
so estimation no longer aborts on large records. Widening the catch so any
estimation error can never fail ingestion is a separate hardening; can follow
up if you prefer.
##########
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:
Moot now. The rework drops the RPC-config read entirely:
`resolveMaxSampleBytes` and the `Network` lookup are gone.
--
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]
