Spark binaryRecords () дает меньшую производительность по сравнению с textFile ()

Question

У меня есть код задания зажигания как показано ниже Который отлично работает с приведенной ниже конфигурацией на кластере.

String path = "/tmp/one.txt";
JavaRDD<SomeClass> jRDD = spark.read()
                            .textFile(path)
                            .javaRDD()
                            .map(line -> {
                                return new SomeClass(line);
                            }); 

Dataset responseSet = sparkSession.createDataFrame(jRDD, SomeClass.class);
responseSet.write()
            .format("text")
            .save(path + "processed");

Принимая во внимание, что если я хочу прочитать двоичный файл (такого же размера, как текст), это займет гораздо больше времени.

String path = "/tmp/one.txt";
JavaRDD<SomeClass> jRDD = sparkContext
                            .binaryRecords(path, 10000, new Configuration())
                            .toJavaRDD()
                            .map(line -> {
                                return new SomeClass(line);
                            }); 

Dataset responseSet = spark.createDataFrame(jRDD, SomeClass.class);
responseSet.write()
            .format("text")
            .save(path + "processed");      

Ниже моя конфигурация.

driver-memory 8g
executor-memory 6g
num-executors 16

Время, затраченное на первый код с файлом 150 МБ, составляет 1,30 минуты. Время, затрачиваемое вторым кодом с файлом размером 150 МБ, составляет 4 минуты.

Кроме того, первый код мог работать на всех 16 исполнителях, тогда как второй использует только один.

Нет предложений, почему это медленно?

Answer 1

Я нашел проблему.Метод textFile() создавал 16 partitions (вы можете проверить numOfPartitions, используя метод getNumPartitions() в RDD), тогда как binaryRecords() создал только 1 (Java binaryRecords не предоставляет перегруженный метод, который указывает количество создаваемых разделов).

Я увеличил numOfPartitions на СДР, созданном binaryRecords() с использованием метода repartition(NUM_OF_PARTITIONS) на СДР.