java - Apache Spark tarda de 5 a 6 minutos en contar de forma sencilla 1 billón de filas de Cassandra

apache-spark spark-cassandra-connector

Question

Estoy usando el conector Spark Cassandra. Se necesitan de 5 a 6 minutos para obtener datos de la tabla de Cassandra. En Spark he visto muchas tareas y Executor en log. ¡La razón podría ser que Spark dividió el proceso en muchas tareas!

A continuación está mi ejemplo de código:

public static void main(String[] args) {

    SparkConf conf = new SparkConf(true).setMaster("local[4]")
            .setAppName("App_Name")
            .set("spark.cassandra.connection.host", "127.0.0.1");

    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD<Demo_Bean> empRDD = javaFunctions(sc).cassandraTable("dev",
            "demo");
    System.out.println("Row Count"+empRDD.count());
}

Answer 1

Después de buscar en Google, me gusta el problema en el último conector spark-cassandra. El parámetro spark.cassandra.input.split.size_in_mb El valor predeterminado es 64 MB que se interpreta como 64 bytes en el código. Así que intente con spark.cassandra.input.split.size_in_mb = 64 * 1024 * 1024 = 67108864

Escuchar es un ejemplo:

public static void main(String[] args) {

    SparkConf conf = new SparkConf(true).setMaster("local[4]")
            .setAppName("App_Name")
            .set("spark.cassandra.connection.host", "127.0.0.1")
            .set("spark.cassandra.input.split.size_in_mb","67108864");


    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD<Demo_Bean> empRDD = javaFunctions(sc).cassandraTable("dev",
            "demo");
    System.out.println("Row Count"+empRDD.count());
}

Answer 2

Para acelerarlo, puede intentar configurar spark.cassandra.input.split.size_in_mb al crear SparkConf.

Podría ser que los ejecutores están tratando de leer todas las filas en la memoria a la vez. Si no todos encajan, podría causar que coloque el RDD en el disco, lo que ocasiona un tiempo lento. Al especificar un tamaño dividido, contará las filas en fragmentos y luego las descartará en lugar de enviarlas al disco.

Puede ver un ejemplo de cómo establecer el tamaño de división aquí .