iteratorSeguro

El problema con toLocalIterator() es que trae todos los datos al driver, y Sonar lo marca como riesgo de memoria si el dataset es grande. La solución es procesar los datos en particiones usando mapPartitions, que mantiene los datos en los nodos del cluster y solo itera localmente por partición. 

n método auxiliar que reemplace toLocalIterator() y devuelva un Iterator de manera segura por partición:

public static <T> Iterator<T> iteratorSeguro(Dataset<T> dataset) {

    return dataset.mapPartitions((Iterator<T> partition) -> {

        List<T> buffer = new ArrayList<>();

        partition.forEachRemaining(buffer::add);

        return buffer.iterator();

    }, Encoders.javaSerialization(Object.class)).toLocalIterator();

}

public static <T> Iterator<T> iteratorSeguro(Dataset<T> dataset) {

    return dataset.mapPartitions((Iterator<T> partition) -> {

        List<T> buffer = new ArrayList<>();

        partition.forEachRemaining(buffer::add);

        return buffer.iterator();

    }, Encoders.javaSerialization(dataset.encoder().clsTag()))

                 .toLocalIterator();

}

Ventajas:

1. Procesa por partición, no carga todo en memoria del driver.

2. Compatible con tu flujo actual de batches.

3. Soluciona la vulnerabilidad marcada por Sonar.

Si quieres, puedo escribir la versión final del método procesarEnBatches usando este iteratorSeguro listo para Spark sin cambios en tu lógica de batching.