/** 
 * Zips this RDD with another one, returning key-value pairs with the first element in each RDD, 
 * second element in each RDD, etc. Assumes that the two RDDs have the *same number of 
 * partitions* and the *same number of elements in each partition* (e.g. one was made through 
 * a map on the other). 
 */ 
  def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)] = withScope { 
  zipPartitions(other, preservesPartitioning = false) { 

 //如果2个迭代器都有值，则输出，如果没值，则不输出，并且两个迭代器的长度必须保持一致

(thisIter, otherIter) => 
    new Iterator[(T, U)] { 
      def hasNext: Boolean = (thisIter.hasNext, otherIter.hasNext) match { 
        case (true, true) => true 
        case (false, false) => false 
        case _ => throw new SparkException("Can only zip RDDs with " + 
          "same number of elements in each partition") 
      } 
      def next(): (T, U) = (thisIter.next(), otherIter.next()) 
    } 
  }

}

def zipPartitions[B: ClassTag, V: ClassTag] 
    (rdd2: RDD[B], preservesPartitioning: Boolean) 
    (f: (Iterator[T], Iterator[B]) => Iterator[V]): RDD[V] = withScope { 
  new ZippedPartitionsRDD2(sc, sc.clean(f), this, rdd2, preservesPartitioning)

}

private[spark] class ZippedPartitionsRDD2[A: ClassTag, B: ClassTag, V: ClassTag]( 
    sc: SparkContext, 
    var f: (Iterator[A], Iterator[B]) => Iterator[V], 
    var rdd1: RDD[A], 
    var rdd2: RDD[B], 
    preservesPartitioning: Boolean = false) 
  extends ZippedPartitionsBaseRDD[V](sc, List(rdd1, rdd2), preservesPartitioning) { 
  //compute就是利用zip时候生成的迭代器，重写其hasNext和next方法来返回数据 
  override def compute(s: Partition, context: TaskContext): Iterator[V] = { 
    val partitions = s.asInstanceOf[ZippedPartitionsPartition].partitions 
    f(rdd1.iterator(partitions(0), context), rdd2.iterator(partitions(1), context)) 
  } 
   
  override def clearDependencies() { 
    super.clearDependencies() 
    rdd1 = null 
    rdd2 = null 
    f = null 
  } 
}

private[spark] abstract class ZippedPartitionsBaseRDD[V: ClassTag]( 
    sc: SparkContext, 
    var rdds: Seq[RDD[_]], 
    preservesPartitioning: Boolean = false) 
  extends RDD[V](sc, rdds.map(x => new OneToOneDependency(x))) { 
    //由于zip默认的preservesPartitioning为false，则zip之后partitioner为None 
  override val partitioner = 
    if (preservesPartitioning) firstParent[Any].partitioner else None 
   
  override def getPartitions: Array[Partition] = { 
    val numParts = rdds.head.partitions.length

//zip的左右两RDD的分区个数必须保持一致 
    if (!rdds.forall(rdd => rdd.partitions.length == numParts)) { 
      throw new IllegalArgumentException("Can't zip RDDs with unequal numbers of partitions") 
    } 
    Array.tabulate[Partition](numParts) { i =>

//获取每个分区的loc信息 
      val prefs = rdds.map(rdd => rdd.preferredLocations(rdd.partitions(i))) 
      // Check whether there are any hosts that match all RDDs; otherwise return the union

//求loc的交集 
      val exactMatchLocations = prefs.reduce((x, y) => x.intersect(y))

//如果交集非空，则其交集为zip之后的loc，如果交集为空，则取两者的非重合的loc 
      val locs = if (!exactMatchLocations.isEmpty) exactMatchLocations else prefs.flatten.distinct 
      new ZippedPartitionsPartition(i, rdds, locs) 
    } 
  } 
   
  override def getPreferredLocations(s: Partition): Seq[String] = { 
    s.asInstanceOf[ZippedPartitionsPartition].preferredLocations 
  } 
   
  override def clearDependencies() { 
    super.clearDependencies() 
    rdds = null 
  }

}

def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag] 
    (rdd2: RDD[B], rdd3: RDD[C], preservesPartitioning: Boolean) 
    (f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V] = withScope { 
  new ZippedPartitionsRDD3(sc, sc.clean(f), this, rdd2, rdd3, preservesPartitioning) 
} 
   
  def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag] 
    (rdd2: RDD[B], rdd3: RDD[C]) 
    (f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V] = withScope { 
  zipPartitions(rdd2, rdd3, preservesPartitioning = false)(f) 
} 
   
  def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag] 
    (rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D], preservesPartitioning: Boolean) 
    (f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V] = withScope { 
  new ZippedPartitionsRDD4(sc, sc.clean(f), this, rdd2, rdd3, rdd4, preservesPartitioning) 
} 
   
  def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag] 
    (rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D]) 
    (f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V] = withScope { 
  zipPartitions(rdd2, rdd3, rdd4, preservesPartitioning = false)(f) 
}

def zipWithIndex(): RDD[(T, Long)] = withScope { 
  new ZippedWithIndexRDD(this)

}

class ZippedWithIndexRDD[T: ClassTag](@transient prev: RDD[T]) extends RDD[(T, Long)](prev) { 
   
  /** The start index of each partition. */ 
  @transient private val startIndices: Array[Long] = { 
    val n = prev.partitions.length 
    if (n == 0) { 
      Array[Long]() 
    } else if (n == 1) { 
      Array(0L) 
    } else {

//先计算出每个分区的起始偏移量，例如假设有4个分区，每个分区依次包含3,4,3个元素，则startIndices为：[0,3,7] 
      prev.context.runJob( 
        prev, 
        Utils.getIteratorSize _, 
        0 until n - 1, // do not need to count the last partition 
        allowLocal = false 
      ).scanLeft(0L)(_ + _) 
    } 
  } 
   
  override def getPartitions: Array[Partition] = {

//组装分区的配置信息，即该分区的对应起始偏移量为startIndices(x.index) 
    firstParent[T].partitions.map(x => new ZippedWithIndexRDDPartition(x, startIndices(x.index))) 
  } 
   
  override def getPreferredLocations(split: Partition): Seq[String] = 
    firstParent[T].preferredLocations(split.asInstanceOf[ZippedWithIndexRDDPartition].prev) 
   
  override def compute(splitIn: Partition, context: TaskContext): Iterator[(T, Long)] = { 
    val split = splitIn.asInstanceOf[ZippedWithIndexRDDPartition]

//通过zipWithIndex计算每个元素在自己分区内的相对偏移量，然后叠加split.startIndex计算在整个RDD中的绝对偏移量 
    firstParent[T].iterator(split.prev, context).zipWithIndex.map { x => 
      (x._1, split.startIndex + x._2) 
    } 
  } 
}

scala> var rdd2 = sc.makeRDD(Seq("A","B","R","D","F"),2)

rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[34] at makeRDD at :21

scala> rdd2.zipWithIndex().collect

res27: Array[(String, Long)] = Array((A,0), (B,1), (R,2), (D,3), (F,4))

/** 
 * Zips this RDD with generated unique Long ids. Items in the kth partition will get ids k, n+k, 
 * 2*n+k, ..., where n is the number of partitions. So there may exist gaps, but this method 
 * won't trigger a spark job, which is different from [[org.apache.spark.rdd.RDD#zipWithIndex]]. 
 * 
 * Note that some RDDs, such as those returned by groupBy(), do not guarantee order of 
 * elements in a partition. The unique ID assigned to each element is therefore not guaranteed, 
 * and may even change if the RDD is reevaluated. If a fixed ordering is required to guarantee 
 * the same index assignments, you should sort the RDD with sortByKey() or save it to a file. 
 */ 
  def zipWithUniqueId(): RDD[(T, Long)] = withScope { 
  val n = this.partitions.length.toLong

//k为分区索引 
  this.mapPartitionsWithIndex { case (k, iter) =>

//i为每个分区元素在自己分区的索引 
    iter.zipWithIndex.map { case (item, i) =>

//i*n+k:基准值为该分区的索引，每次增加分区数目*每个分区元素在自己分区的索引的距离 
      (item, i * n + k) 
    } 
  }

}

scala> var rdd1 = sc.makeRDD(Seq("A","B","C","D","E","F"),2)

rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[44] at makeRDD at :21

//rdd1有两个分区，

scala> rdd1.zipWithUniqueId().collect

res32: Array[(String, Long)] = Array((A,0), (B,2), (C,4), (D,1), (E,3), (F,5))

/** 
 * Applies a function f to all elements of this RDD. 
 */ 
  def foreach(f: T => Unit): Unit = withScope { 
  val cleanF = sc.clean(f) 
  sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF)) 
}

scala> var cnt = sc.accumulator(0)

cnt: org.apache.spark.Accumulator[Int] = 0

scala> var rdd1 = sc.makeRDD(1 to 10,2)

rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at makeRDD at :21

scala> rdd1.foreach(x => cnt += x)

scala> cnt.value

res51: Int = 55