Spark中的各種action算子操作（java版）

在我看來，Spark編程中的action算子的作用就像一個觸發器，用來觸發之前的transformation算子。transformation操作具有懶加載的特性，你定義完操作之后並不會立即加載，只有當某個action的算子執行之后，前面所有的transformation算子才會全部執行。常用的action算子如下代碼所列：（java版） 
package cn.spark.study.core;

import java.util.Arrays; 
import java.util.List; 
import java.util.Map;

import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.JavaPairRDD; 
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;

import scala.Tuple2;

/** 
* action操作實戰 
* @author dd 
* 
*/ 
public class ActionOperation { 
public static void main(String[] args) { 
//reduceTest(); 
//collectTest(); 
//countTest(); 
//takeTest(); 
countByKeyTest(); 
}

 

/**
 * reduce算子
 * 案例：求累加和
 */
private static void reduceTest(){
    SparkConf conf = new SparkConf()
                    .setAppName("reduce")
                    .setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9,10);

    JavaRDD<Integer> numbersRDD = sc.parallelize(numberList);

    //使用reduce操作對集合中的數字進行累加
    int sum = numbersRDD.reduce(new Function2<Integer, Integer, Integer>() {

        @Override
        public Integer call(Integer arg0, Integer arg1) throws Exception {
            return arg0+arg1;
        }
    });

    System.out.println(sum);

    sc.close();
}

/**
 * collect算子
 * 可以將集群上的數據拉取到本地進行遍歷（不推薦使用）
 */
private static void collectTest(){
    SparkConf conf = new SparkConf()
    .setAppName("collect")
    .setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9,10);

    JavaRDD<Integer> numbersRDD = sc.parallelize(numberList);

    JavaRDD<Integer> doubleNumbers = numbersRDD.map(new Function<Integer, Integer>() {

        @Override
        public Integer call(Integer arg0) throws Exception {
            // TODO Auto-generated method stub
            return arg0*2;
        }
    });

    //foreach的action操作是在遠程集群上遍歷rdd中的元素，而collect操作是將在分布式集群上的rdd
    //數據拉取到本地，這種方式一般不建議使用，因為如果rdd中的數據量較大的話，比如超過一萬條，那么性能會
    //比較差，因為要從遠程走大量的網絡傳輸，將數據獲取到本地，有時還可能發生oom異常，內存溢出
    //所以還是推薦使用foreach操作來對最終的rdd進行處理
    List<Integer> doubleNumList = doubleNumbers.collect();
    for(Integer num : doubleNumList){
        System.out.println(num);
    }
    sc.close();
}

/**
 * count算子
 * 可以統計rdd中的元素個數
 */
private static void countTest(){
    SparkConf conf = new SparkConf()
    .setAppName("count")
    .setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9,10);

    JavaRDD<Integer> numbersRDD = sc.parallelize(numberList);

    //對rdd使用count操作統計rdd中元素的個數
    long count = numbersRDD.count();
    System.out.println(count);

    sc.close();
}

/**
 * take算子
 * 將遠程rdd的前n個數據拉取到本地
 */
private static void takeTest(){
    SparkConf conf = new SparkConf()
    .setAppName("take")
    .setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9,10);

    JavaRDD<Integer> numbersRDD = sc.parallelize(numberList);

    //take操作與collect操作類似，也是從遠程集群上獲取rdd數據，但是，collect操作獲取的是rdd的
    //所有數據，take獲取的只是前n個數據
    List<Integer> top3number = numbersRDD.take(3);
    for(Integer num : top3number){
        System.out.println(num);
    }
    sc.close();
}

/**
 * saveAsTextFile算子
 * 
 */
private static void saveAsTExtFileTest(){
    SparkConf conf = new SparkConf()
    .setAppName("saveAsTextFile");

    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9,10);

    JavaRDD<Integer> numbersRDD = sc.parallelize(numberList);

    JavaRDD<Integer> doubleNumbers = numbersRDD.map(new Function<Integer, Integer>() {

        @Override
        public Integer call(Integer arg0) throws Exception {
            // TODO Auto-generated method stub
            return arg0*2;
        }
    });

    //saveAsTextFile算子可以直接將rdd中的數據保存在hdfs中
    //但是我們在這里只能指定保存的文件夾也就是目錄，那么實際上，會保存為目錄中的
    //  /double_number.txt/part-00000文件
    doubleNumbers.saveAsTextFile("hdfs://spark1:9000/double_number.txt");

    sc.close();
}

/**
 * countByKey算子
 */

private static void countByKeyTest(){
    SparkConf conf = new SparkConf()
    .setAppName("take")
    .setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    List<Tuple2<String, String>> studentsList = Arrays.asList(
            new Tuple2<String, String>("class1","leo"),
            new Tuple2<String, String>("class2","jack"),
            new Tuple2<String, String>("class1","marry"),
            new Tuple2<String, String>("class2","tom"),
            new Tuple2<String, String>("class2","david"));

    JavaPairRDD<String, String> studentsRDD = sc.parallelizePairs(studentsList);

    //countByKey算子可以統計每個key對應元素的個數
    //countByKey返回的類型直接就是Map<String,Object>

    Map<String, Object> studentsCounts = studentsRDD.countByKey();

    for(Map.Entry<String, Object> studentsCount : studentsCounts.entrySet()){
        System.out.println(studentsCount.getKey()+" : "+studentsCount.getValue());
    }
    sc.close();
}

 

原文引自：http://blog.csdn.net/kongshuchen/article/details/51344124