hudi-agent commented on code in PR #19550:
URL: https://github.com/apache/hudi/pull/19550#discussion_r3738086155
##########
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());
+ return (long) (rpcMaxSizeMb * BYTES_PER_MB *
SAMPLE_WRITES_TASK_BYTES_FRACTION);
+ }
+
+ /**
+ * Takes up to {@code maxCount} records from {@code sourceRecords}, stopping
early once the
+ * accumulated serialized size would exceed {@code maxBytes}. Each sampled
record is re-keyed with
+ * an empty partition path so the whole sample writes to a single
non-partitioned file, instead of
+ * fanning out into one tiny file per source partition and skewing the
estimate. At least one
+ * record is always retained so a single oversized record still yields an
estimate.
+ *
+ * @param sourceRecords the source records to sample from
+ * @param maxCount the maximum number of records to sample
+ * @param maxBytes the maximum total serialized size (in bytes) of the
sampled records
+ * @return an iterator over the bounded sample
+ */
+ static Iterator<HoodieRecord> takeBoundedSample(Iterator<HoodieRecord>
sourceRecords, int maxCount, long maxBytes) throws IOException {
+ List<HoodieRecord> samples = new ArrayList<>();
+ long accumulatedBytes = 0L;
+ while (sourceRecords.hasNext() && samples.size() < maxCount) {
+ HoodieRecord source = sourceRecords.next();
+ HoodieRecord sample = source.newInstance(new
HoodieKey(source.getRecordKey(), ""));
+ long recordBytes = SerializationUtils.serialize(sample).length;
+ if (!samples.isEmpty() && accumulatedBytes + recordBytes > maxBytes) {
+ break;
+ }
+ samples.add(sample);
+ accumulatedBytes += recordBytes;
Review Comment:
🤖 nit: the `""` literal carries the semantics "write to a single
non-partitioned file", which the Javadoc explains but the call-site doesn't.
Could you pull it out as a class-level constant (e.g.
`SAMPLE_WRITES_PARTITION_PATH = ""`) so both this line and the test's
`serializedSizeOfFirst` helper point to the same named thing rather than two
independent empty strings?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
