[HUDI-7508] Avoid converting iterator to list HoodieStreamerUtils.createHoodieRecords

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 0.15.0, 1.0.0
• Component/s: deltastreamer
• Labels: pull-request-available
• Story Points: 4
• Assignee: Vinish Reddy
• Reporter: Vinish Reddy
• Created: 15/Mar/24 17:39
• Updated: 15/May/24 14:20
• Resolved: 15/May/24 14:20

Description

This block of code is problematic and can lead to OOM when we are we converting the iterator into a list and then returning the iterator back. This just holds up memory in the heap when the executor is running this block of code.
https://github.com/apache/hudi/blob/master/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java#L86

records = avroRDD.mapPartitions(
(FlatMapFunction<Iterator<GenericRecord>, Either<HoodieRecord,String>>) genericRecordIterator -> {
if (autoGenerateRecordKeys)

{{

{ props.setProperty(KeyGenUtils.RECORD_KEY_GEN_PARTITION_ID_CONFIG, String.valueOf(TaskContext.getPartitionId())); props.setProperty(KeyGenUtils.RECORD_KEY_GEN_INSTANT_TIME_CONFIG, instantTime); }

}}

BuiltinKeyGenerator builtinKeyGenerator = (BuiltinKeyGenerator) HoodieSparkKeyGeneratorFactory.createKeyGenerator(props);
List<Either<HoodieRecord,String>> avroRecords = new ArrayList<>();
while (genericRecordIterator.hasNext()) {
GenericRecord genRec = genericRecordIterator.next();
try

{{

{ HoodieKey hoodieKey = new HoodieKey(builtinKeyGenerator.getRecordKey(genRec), builtinKeyGenerator.getPartitionPath(genRec)); GenericRecord gr = isDropPartitionColumns(props) ? HoodieAvroUtils.removeFields(genRec, partitionColumns) : genRec; HoodieRecordPayload payload = shouldCombine ? DataSourceUtils.createPayload(cfg.payloadClassName, gr, (Comparable) HoodieAvroUtils.getNestedFieldVal(gr, cfg.sourceOrderingField, false, props.getBoolean( KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.key(), Boolean.parseBoolean(KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue())))) : DataSourceUtils.createPayload(cfg.payloadClassName, gr); avroRecords.add(Either.left(new HoodieAvroRecord<>(hoodieKey, payload))); }

}}

catch (Exception e) {
if (!shouldErrorTable)

{{

{ throw e; }

}}

avroRecords.add(generateErrorRecord(genRec));
}
}
return avroRecords.iterator();
});
 
 

Issue Links

• links to: GitHub Pull Request #10872