欢迎光临
Spark-Transformation以及Action开发实战
   

Spark-Transformation以及Action开发实战

文章目录

    • 创建RDD
    • Transformation以及Action
    • Transformation开发
    • Action开发
    • RDD持久化
    • 共享变量

创建RDD

  • RDD是Spark的编程核心,在进行Spark编程是,首要任务就是创建一个初始的RDD
  • Spark提供三种创建RDD方式:集合、本地文件、HDFS文件
    • 集合:主要用于本地测试,在实际部署到集群运行之前,自己使用集合构造测试数据,测试Spark流程
    • 本地文件:临时性的处理工作
    • HDFS:最常用的生产上的方式

使用集合创建RDD

  • 通过SparkContext.parallelize方法将集合转化为RDD
  • 通过parallelize方法可以设置RDD的partition数量,Spark会为每一个partition运行一个task来处理
    public static void main(String[] args) {
   
        SparkConf sparkConf = new SparkConf();
        sparkConf.setAppName("CreateRDDArrayJava").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        List list = Arrays.asList(1, 2, 3, 4);
        JavaRDD rdd = sc.parallelize(list, 2);
        Integer sum = rdd.reduce(new Function2() {
   
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
   
                return v1 + v2;
            }
        });
        System.out.println("sum:" + sum);
    }
  def main(args: Array[String]): Unit = {
   
    val conf = new SparkConf();
    conf.setAppName("CreateRDDArray").setMaster("local")
    val context = new SparkContext(conf);
    val arr = Array(1,2,3,4)
    // 集合创建RDD
    val rdd = context.parallelize(arr, 2)
    val sum = rdd.reduce(_ + _)
    println("sum=:" + sum)
  }
  

Spark-Transformation以及Action开发实战,在这里插入图片描述,第1张

使用本地文件以及HDFS文件创建RDD

  • 通过SparkContext.textFile方法创建RDD,这时的RDD就是一行一行的文件数据
  • textFile方法支持针对目录、压缩文件以及通配符的方式
  • 默认会为HDFS文件的每一个Block创建一个partition,也可以通过textFile手动设置分区数量,只能比Block多,不能比Block少

这个可以参考上一篇blog(文件路径可以是hdfs://hadoop01:9000/test/hello,也可以是本地路径):https://blog.csdn.net/Grady00/article/details/136736362

Transformation以及Action

  • Spark支持两种RDD操作:Transformation、Action
    • Transformation可以理解为转换的意思,表示针对RDD数据的一个转换操作,主要对已有的RDD创建一个新的RDD,常见的有MAP,flatMap,filter等。
    • Transformation的特性:lazy,如果一个Spark任务只定义了Transformation算子,即使执行这个任务,任务中的算子也不会真正执行,也就是Transformation算子是不会出发Spark任务执行的,只是记录了对RDD的一些操作,只有进行了Action操作之后所有的Transformation才会真正执行。Spark通过lazy这种特性,来执行底层的Spark任务执行的优化,避免产生过多的中间结果
    • Action可以理解为执行,出发任务执行的操作,主要是对RDD进行最后的操作,比如遍历、reduce、保存到文件等,还可以把结果返回给Driver
    • Action特性:执行Action操作才会出发一个Spark Job的运行,从而触发这个Action之前所有的Transformation操作

不管是Transformation还是Action中的操作,我们都把它称为算子,比如map算子,reduce算子等等

Transformation开发

public static void main(String[] args) {
   
        SparkConf sparkConf = new SparkConf();
        sparkConf.setAppName("TransformationJava").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
		
    // 将RDD中的每个元素进行处理,一进一出
        mapOp(sc);
    // 对RDD的每个元素进行判断,返回true则保存
        filterOp(sc);
    // 与map类似,但是每个元素都可以返回一个或多个新元素
        flatMapOp(sc);
    // 根据key进行分组,每个key对应一个Iterable
    //BN:15003 15005 
	//US:150001 15002 
	//IN:15004 	
        groupByKeyOp(sc);
        groupByKeyOp2(sc);
    // 对每个相同的key对应的value进行reduce操作
    //reduceByKeyOp result:(BN,2)
	//reduceByKeyOp result:(US,2)
	//reduceByKeyOp result:(IN,1)
        reduceByKeyOp(sc);
    // 对每个相同的key对应的value进行排序操作
        sortedByKeyOp(sc);
    // 对两个包含对的RDD进行join操作
        joinOp(sc);
    // 对RDD中的元素去重
        distinctOp(sc);
    }
    /**
     * 分组
     * @param sc
     */
    private static void groupByKeyOp(JavaSparkContext sc) {
   
        Tuple2 t1 = new Tuple2<>(150001, "US");
        Tuple2 t2 = new Tuple2<>(15002, "US");
        Tuple2 t3 = new Tuple2<>(15003, "BN");
        Tuple2 t4 = new Tuple2<>(15004, "IN");
        Tuple2 t5 = new Tuple2<>(15005, "BN");
        List> list = Arrays.asList(t1, t2, t3, t4, t5);
        JavaRDD> rdd = sc.parallelize(list);
        rdd.mapToPair(new PairFunction, String, Integer>() {
   
            @Override
            public Tuple2 call(Tuple2 tup) throws Exception {
   
                return new Tuple2<>(tup._2, tup._1);
            }
        }).groupByKey().foreach(new VoidFunction>>() {
   
            @Override
            public void call(Tuple2> tup) throws Exception {
   
                String area = tup._1;
                System.out.print(area + ":");
                Iterable id = tup._2;
                for (Integer item: id
                     ) {
   
                    System.out.print(item + " ");
                }
                System.out.println();
            }
        });
    }
    /**
     * 拆分
     * @param sc
     */
    private static void flatMapOp(JavaSparkContext sc) {
   
        JavaRDD javaRDD = sc.parallelize(Arrays.asList("good work", "work hard", "tom good", "take it easy"));
        javaRDD.flatMap(new FlatMapFunction() {
   
            @Override
            public Iterator call(String line) throws Exception {
   
                String[] words = line.split(" ");
                return Arrays.asList(words).iterator();
            }
        }).
 
 带一的成语  冰川时代2  三皇五帝是什么意思  石潭积黛色  个性网情侣网名  盆栽水果 
打赏
版权声明:本文采用知识共享 署名4.0国际许可协议 [BY-NC-SA] 进行授权
文章名称:《Spark-Transformation以及Action开发实战》
文章链接:https://goodmancom.com/wl/176013.html