Spark unpersist（）有不同的策略吗？

问题描述：

刚做了一些关于spark unpersist（）的实验，并对实际做了什么感到困惑。我搜索了很多，几乎所有的人都说unpersist（）会立即从excutor的内存中驱逐RDD。但在这个测试中，我们可以看到它并不总是真实的。请参见下面的简单的测试：Spark unpersist（）有不同的策略吗？

private static int base = 0; 
public static Integer[] getInts(){ 
    Integer[] res = new Integer[5]; 
    for(int i=0;i<5;i++){ 
     res[i] = base++; 
    } 
    System.out.println("number generated:" + res[0] + " to " + res[4] + "---------------------------------"); 
    return res; 
} 

public static void main(String[] args) 
{ 
    SparkSession sparkSession = SparkSession.builder().appName("spark test").getOrCreate(); 
    JavaSparkContext spark = new JavaSparkContext(sparkSession.sparkContext()); 

    JavaRDD<Integer> first = spark.parallelize(Arrays.asList(getInts())); 
    System.out.println("first: " + Arrays.toString(first.collect().toArray())); // action 
    first.unpersist(); 
    System.out.println("first is unpersisted"); 

    System.out.println("compute second ========================"); 
    JavaRDD<Integer> second = first.map(i -> { 
     System.out.println("double " + i); 
     return i*2; 
    }).cache(); // transform 
    System.out.println("second: " + Arrays.toString(second.collect().toArray())); // action 
    second.unpersist(); 

    System.out.println("compute third ========================"); 
    JavaRDD<Integer> third = second.map(i -> i+100); // transform 
    System.out.println("third: " + Arrays.toString(third.collect().toArray())); // action 
}

输出为：

number generated:0 to 4--------------------------------- 
first: [0, 1, 2, 3, 4] 
first is unpersisted 
compute second ======================== 
double 0 
double 1 
double 2 
double 3 
double 4 
second: [0, 2, 4, 6, 8] 
compute third ======================== 
double 0 
double 1 
double 2 
double 3 
double 4 
third: [100, 102, 104, 106, 108]

正如我们所看到的，unpersist（） '第一' 是无用的，它不会重新计算。但unpersist（）'second'将触发重新计算。任何人都可以帮我弄清楚为什么unpersist（）'第一个'不会触发重新计算？如果我想强迫'第一'被驱逐出内存，我该怎么办？并行或textFile（）API的RDD有什么特别之处吗？谢谢！

答

此行为与缓存无关，并且unpersisting。实际上first甚至不是persisted，虽然在这里没有太大的区别。

当你parallelize，你通过一个本地，非分布式的对象。 parallelize的参数的值为，其生命周期完全超出了Spark的范围。因此，一旦ParallelCollectionRDD已经初始化，Spark根本没有理由重新计算它。如果你想分发不同的集合，只需创建一个新的RDD。

还值得注意的是，unpersist可以在阻塞和非阻塞模式下调用，具体取决于blocking参数。

Spark unpersist（）有不同的策略吗？

相关推荐