如图所示,测试Spark0与Stage1之间运行顺序与Stage0内pipeline①②的运算顺序,Stage0两个分区,Stage1三个分区;
打印结果在stage0第一个pipeline①执行完后,调用了Stage1内的RDD4,后继续运行Stage0内的pipeline②
如何解释Stage运行时跨分区的情况呢?
object Hello15Pipeline {
def main(args: Array[String]): Unit = {
//初始化配置
val context: SparkContext = new SparkContext(new SparkConf().setMaster("local").setAppName("PPLine"))
//rdd: 构建RDD
val rdd: RDD[Int] = context.parallelize(Array[Int](1,2,3,4,5,6,7,8), 2)
//rdd1:使用map,打印调用记录
val rdd1: RDD[Int] = rdd.map((x) => {
println("rdd1----pid:" + TaskContext.getPartitionId() + "---" + x)
x
})
//rdd2:使用filter,打印调用记录
val rdd2: RDD[Int] = rdd1.filter((x) => {
println("rdd2----pid:" + TaskContext.getPartitionId() + "---" + x)
true
})
//rdd3:使用map,构建元组
val rdd3: RDD[(String, Int)] = rdd2.map((x) => {
println("rdd3----pid:" + TaskContext.getPartitionId() + "---" + x)
Tuple2("测试" + x % 3, x)
})
//rdd4:使用reduceByKey,进行合并,使用shuffle
val rdd4: RDD[(String, Int)] = rdd3.reduceByKey((sum: Int, value: Int) => {
println("rdd4----pid:" + TaskContext.getPartitionId() + "---"+sum +"-----" + "值"+ value)
sum + value
},4)
//rdd5:使用map,打印rdd4运算的结果
val rdd5: RDD[(String, Int)] = rdd4.map((x) => {
println("rdd5----pid:" + TaskContext.getPartitionId() + "---" + x)
x
})
rdd5.count()
context.stop()
}
}
打印结果
rdd1----pid:0---1
rdd2----pid:0---1
rdd3----pid:0---1
rdd1----pid:0---2
rdd2----pid:0---2
rdd3----pid:0---2
rdd1----pid:0---3
rdd2----pid:0---3
rdd3----pid:0---3
rdd1----pid:0---4
rdd2----pid:0---4
rdd3----pid:0---4
rdd4----pid:0---1-----值4
rdd1----pid:1---5
rdd2----pid:1---5
rdd3----pid:1---5
rdd1----pid:1---6
rdd2----pid:1---6
rdd3----pid:1---6
rdd1----pid:1---7
rdd2----pid:1---7
rdd3----pid:1---7
rdd1----pid:1---8
rdd2----pid:1---8
rdd3----pid:1---8
rdd4----pid:1---5-----值8
rdd4----pid:0---2-----值13
rdd5----pid:0---(测试2,15)
rdd4----pid:2---3-----值6
rdd5----pid:2---(测试0,9)
rdd4----pid:3---5-----值7
rdd5----pid:3---(测试1,12)
Process finished with exit code 0