foreach

---

官方文檔描述：

Applies a function f to all elements of this RDD.

函數原型：

def foreach(f: VoidFunction[T])

foreach用于遍歷RDD,將函數f應用于每一個元素。

源碼分析：

def foreach(f: T => Unit): Unit = withScope { val cleanF = sc.clean(f) sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF)) }

實例：

List<Integer> data = Arrays.asList(5, 1, 1, 4, 4, 2, 2); JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data,3); javaRDD.foreach(new VoidFunction<Integer>() { @Override public void call(Integer integer) throws Exception { System.out.println(integer); } });

foreachPartition

---

官方文檔描述：

Applies a function f to each partition of this RDD.

函數原型：

def foreachPartition(f: VoidFunction[java.util.Iterator[T]])

foreachPartition和foreach類似，只不過是對每一個分區使用f。

源碼分析：

def foreachPartition(f: Iterator[T] => Unit): Unit = withScope { val cleanF = sc.clean(f) sc.runJob(this, (iter: Iterator[T]) => cleanF(iter)) }

實例：

List<Integer> data = Arrays.asList(5, 1, 1, 4, 4, 2, 2); JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data,3);//獲得分區ID JavaRDD<String> partitionRDD = javaRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<Integer>, Iterator<String>>() { @Override public Iterator<String> call(Integer v1, Iterator<Integer> v2) throws Exception { LinkedList<String> linkedList = new LinkedList<String>(); while(v2.hasNext()){ linkedList.add(v1 + "=" + v2.next()); }return linkedList.iterator(); } },false); System.out.println(partitionRDD.collect()); javaRDD.foreachPartition(new VoidFunction<Iterator<Integer>>() { @Override public void call(Iterator<Integer> integerIterator) throws Exception { System.out.println("___________begin_______________"); while(integerIterator.hasNext()) System.out.print(integerIterator.next() + " "); System.out.println("\n___________end_________________"); } });

lookup

---

官方文檔描述：

Return the list of values in the RDD for key `key`. This operation is done efficiently if the RDD has a known partitioner by only searching the partition that the key maps to.

函數原型：

def lookup(key: K): JList[V]

• ?

lookup用于(K,V)類型的RDD,指定K值，返回RDD中該K對應的所有V值。

源碼分析：

def lookup(key: K): Seq[V] = self.withScope { self.partitioner match { case Some(p) => val index = p.getPartition(key) val process = (it: Iterator[(K, V)]) => { val buf = new ArrayBuffer[V] for (pair <- it if pair._1 == key) { buf += pair._2 } buf } : Seq[V] val res = self.context.runJob(self, process, Array(index), false) res(0) case None => self.filter(_._1 == key).map(_._2).collect() } }

從源碼中可以看出，如果partitioner不為空，計算key得到對應的partition，在從該partition中獲得key對應的所有value；如果partitioner為空，則通過filter過濾掉其他不等于key的值，然后將其value輸出。

實例：

List<Integer> data = Arrays.asList(5, 1, 1, 4, 4, 2, 2); JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data, 3); JavaPairRDD<Integer,Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() { int i = 0; @Override public Tuple2<Integer, Integer> call(Integer integer) throws Exception { i++; return new Tuple2<Integer, Integer>(integer,i + integer); } }); System.out.println(javaPairRDD.collect()); System.out.println("lookup------------" + javaPairRDD.lookup(4));

總結

以上是生活随笔為你收集整理的Spark Java API：foreach、foreachPartition、lookup的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。