Почему Spark Core тормозит?

При работе только с spark sql запросы на базу обрабатываются очень быстро, но при подключении JavaPairJDD он начинает тормозить.

Я хотел бы узнать, есть ли возможность ускорить обработку данных или же как-то улучшить код для того чтобы он не зависал?

Вот запуск без JavaPairRDD

collectAsList at StackOverFlow.java:56, took 0.883860 s

А вот с JavaPairRDD

collectAsMap at StackOverFlow.java:61, took 128.340516 s

А вот сам Код

public List<Order> getAllWithoutPairRDD(Optional<String> search, Optional<Integer> size) {

    SparkSession session = SparkSession.builder().config(config).getOrCreate();

    Properties properties = new Properties();

    properties.setProperty("partitionColumn", "id");
    properties.setProperty("fetchsize", "1000");
    properties.setProperty("driver", "org.postgresql.Driver");
    properties.setProperty("user", "postgres");
    properties.setProperty("password", "password");

    Dataset<Row> jdbc = session.read().jdbc("jdbc:postgresql://localhost/orders"
            , "orders"
            , "id"
            , 1L
            , 60000000L
            , 100
            , properties
    );
    Dataset<Row> ordersData = jdbc.select(col("*")).where(col("city_id").equalTo(3L)).limit(size.orElse(1));
    JavaPairRDD<Timestamp, Integer> analyticPairRDD = ordersData.toJavaRDD().mapToPair((PairFunction<Row, Timestamp, Integer>) row -> new Tuple2<Timestamp, Integer>((Timestamp) row.get(0), 1));

    JavaPairRDD<Timestamp,Integer> result = analyticPairRDD.groupByKey().mapValues(Iterables::size);
    return ordersData.as(orderEncoder).collectAsList();
}

public Map<Timestamp, Integer> getAllWithPairRDD(Optional<String> search, Optional<Integer> size) {

    SparkSession session = SparkSession.builder().config(config).getOrCreate();

    Properties properties = new Properties();

    properties.setProperty("partitionColumn", "id");
    properties.setProperty("fetchsize", "1000");
    properties.setProperty("driver", "org.postgresql.Driver");
    properties.setProperty("user", "postgres");
    properties.setProperty("password", "password");

    Dataset<Row> jdbc = session.read().jdbc("jdbc:postgresql://localhost/orders"
            , "orders"
            , "id"
            , 1L
            , 60000000L
            , 100
            , properties
    );
    Dataset<Row> ordersData = jdbc.select(col("operation_date")).where(col("city_id").equalTo(3L)).limit(size.orElse(1));
    JavaPairRDD<Timestamp, Integer> analyticPairRDD = ordersData.toJavaRDD().mapToPair((PairFunction<Row, Timestamp, Integer>) row -> new Tuple2<Timestamp, Integer>((Timestamp) row.get(0), 1));

    JavaPairRDD<Timestamp,Integer> result = analyticPairRDD.groupByKey().mapValues(Iterables::size);
    return result.collectAsMap();
}

@Bean
public SparkConf sparkConf() {
    return new SparkConf()
            .setAppName(appName)
            .setMaster(masterUri)
            .set("spark.executor.memory", "2g")
            .set("spark.executor.cores", "10")
            .set("spark.executor.memoryOverhead", "1g")
            .set("spark.driver.cores", "10")
            .set("spark.driver.memory", "3g")
            .set("spark.yarn.am.memory", "2g")
            .set("spark.yarn.am.cores", "4")
            .set("spark.sql.shuffle.partitions", "1000")
            .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
}