DataFrameWriter.partitionBy не гарантирует порядок сортировки. ⇐ JAVA

[Anonymous]

У меня есть набор данных со следующей схемой:
(timestamp, partsKey, logValue)
Я хочу, чтобы набор данных был отсортировано по временной метке, но записывайте в файл в следующем каталоге:
outputDir/partitionKey/files
Выходной файл содержит только logValue, то есть временная метка используется
только для сортировки и не используется для вывода.
(К вашему сведению, logValue содержит текстовое представление метки времени, которое не поддается сортировке)
Однако, как упоминалось в проблеме Spark SPARK-44512, DataFrameWriter.partitionBy не гарантирует выходные данные сортируются глобально.
(примечание: я обнаружил, что даже установка spark.sql.optimizer.plannedWrite.enabled=false все равно не гарантирует отсортированный результат в среде с нехваткой памяти)
И разработчики говорят, что DataFrameWriter.partitionBy не гарантирует
отсортированные результаты:
"Хотя я понимаю, что Apache Spark 3.4.0 меняет поведение, подобное приведенному выше, я не думаю, что существует контракт, который Apache Spark Операция partBy сохраняет предыдущий порядок."
Чтобы обойти эту проблему, мне пришлось прибегнуть к созданию формата вывода Hadoop путем расширения org.apache.hadoop.mapred.lib. MultipleTextOutputFormat и выведите файл с помощью saveAsHadoopFile:

Код: Выделить всё

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;

public final class PartitionedMultipleTextOutputFormat
extends MultipleTextOutputFormat {
@SuppressWarnings("MissingJavadocMethod")
public PartitionedMultipleTextOutputFormat() {
super();
}

@Override
protected Object generateActualKey(final K key, final V value) {
return null;
}

@Override
protected String generateFileNameForKeyValue(final K key, final V
value, final String leaf) {
return new Path(key.toString(), leaf).toString();
}
}

Код: Выделить всё

private static Tuple2 mapRDDToDomainLogPair(final Row row) {
final String domain = row.getAs("partitionKey");
final var log = (String) row.getAs("logValue");
final var logTextClass = new Text(log);
return new Tuple2(domain, logTextClass);
}

dataset
.sort("timestamp")
.javaRDD()
.mapToPair(TheClass::mapRDDToDomainLogPair)
.saveAsHadoopFile(outputPath, String.class, Text.class,
PartitionedMultipleTextOutputFormat.class, GzipCodec.class);

Это кажется неуклюжим и медленным, чем непосредственное использование DataFrameWriter.partitionBy. Есть ли лучший способ?

Подробнее здесь: https://stackoverflow.com/questions/786 ... sort-order