Je suis un peu confus par votre code, car il mélange des constructions pré-Spark 2.x comme SparkConf et beaucoup de RDD. Ce n'est pas faux de les utiliser, mais depuis Spark 2.x, les choses sont un peu différentes.

Voici un exemple utilisant SparkSession et dataframes, qui est une version sur-ensemble, plus puissante, du RDD (pour faire les choses courtes).

Dans l'exemple, vous verrez plusieurs façons de faire vos opérations map / reduction, deux avec map / reduction et une avec une syntaxe simple de type SQL.

mapper et réduire avec getAs ()

package net.jgp.books.spark.ch07.lab990_others;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

/**
 * Simple ingestion followed by map and reduce operations.
 * 
 * @author jgp
 */
public class SelfIngestionApp {

  /**
   * main() is your entry point to the application.
   * 
   * @param args
   */
  public static void main(String[] args) {
    SelfIngestionApp app = new SelfIngestionApp();
    app.start();
  }

  /**
   * The processing code.
   */
  private void start() {
    // Creates a session on a local master
    SparkSession spark = SparkSession.builder()
        .appName("Self ingestion")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> df = createDataframe(spark);
    df.show(false);

    // map and reduce with getAs()
    int totalLines = df
        .map(
            (MapFunction<Row, Integer>) row -> row.<Integer>getAs("i"),
            Encoders.INT())
        .reduce((a, b) -> a + b);
    System.out.println(totalLines);

    // map and reduce with getInt()
    totalLines = df
        .map(
            (MapFunction<Row, Integer>) row -> row.getInt(0),
            Encoders.INT())
        .reduce((a, b) -> a + b);
    System.out.println(totalLines);

    // SQL-like
    long totalLinesL = df.selectExpr("sum(*)").first().getLong(0);
    System.out.println(totalLinesL);
  }

  private static Dataset<Row> createDataframe(SparkSession spark) {
    StructType schema = DataTypes.createStructType(new StructField[] {
        DataTypes.createStructField(
            "i",
            DataTypes.IntegerType,
            false) });

    List<Integer> data =
        Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6, 7, 8, 9);
    List<Row> rows = new ArrayList<>();
    for (int i : data) {
      rows.add(RowFactory.create(i));
    }

    return spark.createDataFrame(rows, schema);
  }
}

mapper et réduire avec getInt ()

long totalLinesL = df.selectExpr("sum(*)").first().getLong(0);
System.out.println(totalLinesL);

Similaire à SQL

C'est probablement le plus populaire.

totalLines = df
    .map(
        (MapFunction<Row, Integer>) row -> row.getInt(0),
        Encoders.INT())
    .reduce((a, b) -> a + b);
System.out.println(totalLines);

Exemple complet

int totalLines = df
    .map(
        (MapFunction<Row, Integer>) row -> row.<Integer>getAs("i"),
        Encoders.INT())
    .reduce((a, b) -> a + b);
System.out.println(totalLines);

p >

Vous pouvez essayer d'utiliser le générateur de session Spark, et vous pouvez obtenir le contexte Spark par spark.sparkContext ()

public static SparkSession sparkSession(String master,
                                        String appName) {
return    SparkSession.builder().appName(appName)
                       .master(master)
                       .config("spark.dynamicAllocation.enabled", true)
                       .config("spark.shuffle.service.enabled", true)
                       .config("spark.driver.maxResultSize", "8g")
                       .config("spark.executor.memory", "8g")
                       .config("spark.executor.cores", "4")
                       .config("spark.cores.max", "6")
                       .config("spark.submit.deployMode", "client")
                       .config("spark.network.timeout", "3600s")
                       .config("spark.eventLog.enabled", true)
                       .getOrCreate();
}

Si vous utilisez eclipse, vous pouvez définir Exécuter > Configurations d'exécution ... > Arguments > arguments VM et définir max taille de tas comme -Xmx512m.

Dans l’idée, vous pouvez définir Configurations Exécuter \ Déboguer > Options VM: -Xmx512m

Dans votre code, vous pouvez essayer ce conf.set ("spark.testing.memory", "2147480000")

Exception de mémoire du pilote

Cela se produit lorsque le pilote Spark manque de mémoire. C'est à ce moment que le maître d'application qui lance le pilote dépasse la limite et met fin au processus de fil.

Message d'erreur: Java.lang.OutOfMemoryError

Solution: augmentez la mémoire de vos pilotes en configurant ce qui suit:

    conf spark.driver.memory = <XY>g