apache-spark - 了解 Spark 中的持久性

Question

我对 Spark 还很陌生，并且对它进行了一些尝试，并对它有一些疑问:-

这是我正在尝试的代码:-

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;

/**
 * Computes an approximation to pi
 * Usage: JavaSparkPi [slices]
 */
public final class JavaSparkPi {

    public static void main(String[] args) throws Exception {

        SparkConf sparkConf = new SparkConf().setAppName("JavaSparkPi").setMaster("local[4]");


        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        JavaRDD<String> data = jsc.textFile("/Users/sanchay/Desktop/2017-08-08");


        while(true) {

            Thread.sleep(1000);

                    int count = data.map(s -> s.length()).reduce((a, b) -> (a + b));
            System.out.println("count == " + count);

        }

      //  jsc.stop();

    }
}

1) 由于我当前在本地模式下运行，因此每次迭代所花费的时间都会减少(文件大小约为 150 Mb)。这是因为每次 RDD 都会被缓存。我想知道何时在远程节点集群上运行它，我是否会遇到相同的行为，因为同一节点不一定会获得之前计算的相同分区。

2)此外，我尝试取消保留上面的“数据”变量，但是，我注意到所用时间没有差异，因为从第一次迭代到第二次迭代的时间减少了，暗示某处已缓存某些内容。那么，它是否仍然以某种方式被缓存，或者我在这里遗漏了一些明显的东西。

Answer 1

不，默认情况下它不会persist()，我们必须提及要持久化的RDD，而没有外部方法spark不会持久化任何内容

你可以像这样持久化一个RDD

import org.apache.spark.storage.StorageLevel 

val source = s.map(rec => (rec.split(",")(0), rec)).persist(StorageLevel.MEMORY_ONLY)
 //DISK_ONLY,MEMORY_AND_DISK,MEMORY_AND_DISK_SER etc

默认缓存将在内存中进行，我们可以将其指定为 val rdd1 = a.cache()