【spark-spring boot】学习笔记

🕗 发布于 2024-11-27 14:04 spark spring boot 学习

说明

本文依赖于上一篇文章：【spark学习】 spring boot 整合 spark 的相关内容，请先阅读上一篇文章，将spark需要的环境先配置好，并测试通过之后再进行此文章的阅读和操作。

RDD学习

RDD介绍

想象一下你有一个大大的数据表，里面包含了很多很多的信息。如果你想对这些数据进行操作，比如筛选出符合条件的数据、或者对数据做一些计算，RDD 就是 Spark 用来存储和操作这些数据的一种方式。它有两个基本操作：转换操作（就像是加工数据）和 行动操作（获取结果）

RDD案例

基于集合创建RDD

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    /**
     * 创建 RDD 算子
     */
    @PostConstruct
    public void createRDD() {
        // 从集合创建 RDD
        List<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");
        JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDD
        parallelize.collect().forEach(System.out::println); // 打印
    }
}

输出结果：
在这里插入图片描述

RDD存入外部文件中

将RDD存入外部文件，用于观察文件结果和分区结果。txt文件数等于默认分区数。从结果中看出默认分区为4个。

注意：存放结果的文件夹路径必须没有被创建。

以下案例中，将基于集合生成的RDD保存到saveRddDemo文件夹中。

package www.zhangxiaosan.top.timer;


import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    /**
     * 创建 RDD 算子
     */
    @PostConstruct
    public void createRDD() {
        // 从集合创建 RDD
        List<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");
        
        // parallelize(): 创建RDD，RDD中创建默认分区数
        // parallelize( 元素, 分区数)： 创建RDD，RDD中指定分区数
        JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDD
        
        parallelize.collect().forEach(System.out::println); // 打印

        // 存储的目录文件夹路径。此处为项目中的路径且目录必须为不存在的路径。
        String fileSavePath="I:\\zhang\\SpringBootSparkDemo\\src\\main\\resources\\saveRddDemo";
        parallelize.saveAsTextFile(fileSavePath);// 开始将RDD保存到文件中 
    }
}

运行结果：
在文件夹中存放的是运行程序生成的文件。如下图。
_SUCCESS文件：成功标记
part-XXX 文件：保存的数据文件，结果中有4个文件，说明默认分区为4个。
在这里插入图片描述

转换算子操作

转换算子（Transformation）是指那些返回一个新的 RDD 的操作。转换算子不会立即执行计算，而是构建一个执行计划，只有当行动算子（Action）触发计算时，转换算子的操作才会实际执行。转换算子可以用来处理和转换数据，比如映射、过滤、聚合等。

map 操作

说明

map() 方法传入一个函数作为参数。map() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：map( 函数( RDD元素 ) )。 map() 方法会返回一个新的RDD。

案例

将RDD中的每个元素都拼接上字符串 “ say hi ”

package www.zhangxiaosan.top.timer;


import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    /**
     * 创建 RDD 算子
     */
    @PostConstruct
    public void createRDD() {
        // 从集合创建 RDD
        List<String> lists = Arrays.asList("张三", "李四", "王五");
        JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDD

// lambda表达式传入匿名函数，进行元素拼接字符。
        // item 表示RDD中的元素
        parallelize = parallelize.map(item -> item + " say hi"); 
        
        parallelize.collect().forEach(System.out::println); // 打印
    }
}

输出结果：
在这里插入图片描述

flatMap操作

说明

flatMap() 方法传入一个函数作为参数。flatMap() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：flatMap( 函数( RDD元素 ) )。 flatMap() 方法会返回一个新的RDD。
flatMap() 方法会将 RDD的元素扁平化处理成一个集合。

例如：
假设你有一个箱子，箱子里面放着几个小盒子，每个小盒子里又有一些玩具。flatMap 就是一个工具，它能帮你把每个小盒子里的玩具拿出来，直接放进一个大盒子里，最终把所有玩具放在一个地方。
每个小盒子里的玩具可以不止一个，甚至可能没有玩具（比如有的盒子是空的）。flatMap 会把每个盒子里的玩具都拿出来，放到一个大盒子里，最终得到一个扁平的大盒子，里面是所有玩具。

案例

将集合：[ [1, 2, 3, 4, 5],[hello, spark],[张三, 李四] ]
扁平化处理成：[ 1, 2, 3, 4, 5, hello, spark, 张三, 李四]

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void flatMapDemo() {
    // 声明集合：[ [1, 2, 3, 4, 5],[hello, spark],[张三, 李四] ]
        List<List<String>> arrs = Arrays.asList(
                Arrays.asList("1", "2", "3", "4", "5"),
                Arrays.asList("hello", "spark"),
                Arrays.asList("张三", "李四")
        );
        
        //输出声明的集合呢容
        System.out.println("原始集合打印：");
        arrs.forEach(item -> System.out.print(" " + item));

// 分隔符
        System.out.println();
        System.out.println("-----------");

        System.out.println("flatMap操作后：");
        // 创建集合的RDD
        JavaRDD<List<String>> parallelize = javaSparkContext.parallelize(arrs);

// flatMap操作
        List<String> collect = parallelize.flatMap(i -> i.iterator()).collect(); 

// 打印 flatMap操作 后的集合
        collect.forEach(item -> System.out.print(" " + item));
        System.out.println();
    }
}

filter 操作

说明

filter() 方法传入一个函数作为参数，函数返回值只能为Boolean值。filter() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：filter( 函数( RDD元素 ) )。 filter() 方法会返回一个新的RDD。filter() 将RDD元素中满足条件的元素保留，即函数返回为true的元素；RDD中不满足条件的元素，即函数返回为false的元素过滤。

案例

过滤单数，保留双数


import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void filterDemo() {
    // 声明集合
        List<Integer> arrs = Arrays.asList(-1, 0, 1, 2, 3, 4, 5);

        System.out.println("原始集合打印：");
        arrs.forEach(item -> System.out.print(" " + item));
// 输出过滤前的数据
        System.out.println();
        System.out.println("-----------");

        System.out.println("filter操作后：");
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

// 过滤掉 单数 （取模不等于0的数字为单数，否则为偶数）。item % 2 == 0 的数字为双数，返回有true保留
        List<Integer> collect = parallelize.filter(item -> item % 2 == 0).collect();

// 输出过滤后的数据
        collect.forEach(item -> System.out.print(" " + item));
        System.out.println();
    }
}

运行结果：
在这里插入图片描述

groupBy 操作

说明

将数据按某些条件进行分组。每个组由键值对组成。
groupBy() 常和以下聚合函数一起使用，来对分组数据进行统计分析。
常用的聚合操作包括：

count(): 统计每个组的元素数量
sum(): 计算每个组的元素总和
avg(): 计算每个组的平均值
max()：计算每个组的最大值
min(): 计算每个组的最小值
agg(): 自定义聚合操作，可以结合多个聚合函数

案例

给写生成绩按照分数来分组。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void groupByDemo() {
    // 定义学生成绩数据
        List<Tuple2<String, Integer>> students = Arrays.asList(
                new Tuple2<>("张三", 85),
                new Tuple2<>("李四", 90),
                new Tuple2<>("王五", 85),
                new Tuple2<>("赵六", 95),
                new Tuple2<>("孙七", 90)
        );
        
        // 创建RDD
        JavaRDD<Tuple2<String, Integer>> parallelize = javaSparkContext.parallelize(students);

        // 使用 groupBy 进行分组，按成绩分组
        JavaPairRDD<Object, Iterable<Tuple2<String, Integer>>> integerIterableJavaPairRDD = parallelize.groupBy(tuple -> tuple._2());// 使用 groupBy 按成绩分组。_2表示元组的第二个元素，即分数

        // 打印结果
        integerIterableJavaPairRDD.sortByKey().foreach(item -> System.out.println("成绩：" + item._1() + "，学生：" + item._2()));

    }
}

结果如下：
在这里插入图片描述

distinct 操作

说明

对RDD的元素进行分布式去重。返回新的RDD，新的RDD内的元素不重复出现。

案例

将集合中重复的元素去除。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;
    
    @PostConstruct
    public void distinctDemo() {
        // 声明集合
        List<Integer> arrs = Arrays.asList(1,2,3,4,5,6,1,4,6);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // distinct()方法，用于去除重复元素。
        JavaRDD<Integer> distinct = parallelize.distinct();

        // 打印结果
        distinct.collect().forEach(System.out::println);
    }
}

运行结果：
在这里插入图片描述

sortBy 操作

说明

对RDD的元素进行排序，返回新的RDD。
sortBy() 可传入3个参数：
参数1：函数，每个RDD元素都会传入此函数中，此函数定义排序规则。
参数2：boolean值。定义排序顺序。true为升序，false降序。
参数3：Integer 整数，指定分区数量。

案例

将集合中的元素降序排序。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;
    
    @PostConstruct
    public void sortByDemo() {
    // 创建无序集合
        List<Integer> arrs = Arrays.asList(7,2,5,4,1,9,6,3,8);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // 使用 sortBy 进行排序，按元素值排序，并返回一个RDD。
        JavaRDD<Integer> sortBy = parallelize.sortBy(item -> item, false, 1);

        // 打印结果
        sortBy.collect().forEach(System.out::println);
    }
}

运行结果：
在这里插入图片描述

mapToPair 操作

说明

mapToPair()是将一个普通的 RDD 转换为 JavaPairRDD 的一个方法。 JavaPairRDD 中的每个元素都是一个键值对。
mapToPair对RDD的元素进行映射成一个由键值对组成的 RDD（即映射成JavaPairRDD），即将每个元素转换成一个 Tuple2 对象，其中第一个元素是键（key），第二个元素是值（value）。

案例

将集合 [“张三:85”, “李四:90”, “王五:85”, “赵六:95”, “孙七:90”] 中的学生和成绩按照冒号分隔，形成键值对，姓名为键，值为成绩。

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void mapToPairDemo() {
        // 创建包含学生成绩的列表
        List<String> students = Arrays.asList("张三:85", "李四:90", "王五:85", "赵六:95", "孙七:90");
        
        // 将数据并行化为 RDD
        JavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);

        // 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对
        JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair((PairFunction<String, String, Integer>) s -> {
            // 根据冒号分隔学生姓名和成绩，返回一个 (姓名, 成绩) 的元组
            String[] parts = s.split(":");
            // 返回一个 (姓名, 成绩) 的元组
            return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
        });

        // 打印结果
        studentsPairRDD.collect().forEach(System.out::println);
    }
}

运行结果：
在这里插入图片描述

mapValues操作

说明

mapValues() 是一个用于处理 JavaPairRDD 的方法。它可以对 RDD 中的每个键值对的值（value）进行转换，同时保留原来的键（key）不变。

案例

将集合 [“张三:85”, “李四:90”, “王五:85”, “赵六:95”, “孙七:90”] 中的学生和成绩按照冒号分隔，形成键值对，姓名为键，值为成绩。并将成绩在原分数上+5分。


import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void mapToPairDemo() {
        // 创建包含学生成绩的列表
        List<String> students = Arrays.asList("张三:85", "李四:90", "王五:85", "赵六:95", "孙七:90");

        // 将数据并行化为 RDD
        JavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);

        // 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对
        JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair((PairFunction<String, String, Integer>) s -> {
            // 根据冒号分隔学生姓名和成绩，返回一个 (姓名, 成绩) 的元组
            String[] parts = s.split(":");
            // 返回一个 (姓名, 成绩) 的元组
            return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
        });

        // 打印结果
        System.out.println("原始分数：");
        studentsPairRDD.collect().forEach(System.out::println);

        System.out.println("\n+5分后：");
        // item 表示每个键值对中的值，即分数
        JavaPairRDD<String, Integer> newValeRdd = studentsPairRDD.mapValues(item -> item + 5);

        // 打印结果
        newValeRdd.collect().forEach(System.out::println);
    }
}

运行结果：
在这里插入图片描述

groupByKey操作

说明

groupByKey() 是一个用于处理 JavaPairRDD 的方法。它根据键对数据进行分组，将所有具有相同键的元素聚集到一起，生成一个新的 RDD，其中每个键对应一个包含所有相同键值的集合。
每个键对应的值收集到一个 Iterable 容器中，然后返回一个新的 RDD，其中每个键对应一个包含该键的所有值的 Iterable。

案例

将集合 [“张三:85”, “李四:90”, “陈八:95”, “王五:85”, “黄六:92”] 中的学生和成绩按照冒号分隔，形成键值对，成绩为键，值为姓名。并将键值对中的键进行分类，将相同分数的学生归成一组。

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void groupByKeyDemo(){
        // 创建包含学生成绩的列表
        List<String> students = Arrays.asList("张三:85", "李四:90", "陈八:95", "王五:85", "黄六:92");

        // 将数据并行化为 RDD
        JavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);

        // 使用 mapToPair 将每个元素转换为 (成绩, 姓名) 的键值对
        JavaPairRDD<Integer, String> studentsPairRDD = studentsRDD.mapToPair(s -> {
            String[] parts = s.split(":");
            return new Tuple2<Integer, String>(Integer.parseInt(parts[1]), parts[0]);
        });
        
        // 根据键值对中的键进行分类
        JavaPairRDD<Integer, Iterable<String>> groupedRDD = studentsPairRDD.groupByKey();

// 打印结果
        groupedRDD.collect().forEach(pair -> {
            System.out.println("成绩: " + pair._1() + ", 姓名: " + pair._2());
        });
  }
}

执行结果：
在这里插入图片描述

reduceByKey操作

说明

reduceByKey()用于对 JavaPairRDD 数据进行聚合的一个方法。它根据键对值进行合并，并在分区内进行局部聚合，从而减少了跨节点的数据传输，通常比 groupByKey() 更高效。

案例

计算学生的总分。

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void groupByKeyDemo(){
        // 创建包含学生成绩的列表
        List<String> students = Arrays.asList("张三:85", "李四:90", "张三:95", "王五:85", "李四:92");

        // 将数据并行化为 RDD
        JavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);
        
        // 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对
        JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair(s -> {
            String[] parts = s.split(":");
            return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
        });

        // 使用 groupByKey 按学生姓名进行成绩求和
        JavaPairRDD<String, Integer> groupedRDD = studentsPairRDD.reduceByKey((a, b) -> a + b);

        // 打印结果
        groupedRDD.collect().forEach(pair -> {
            System.out.println("姓名: " + pair._1() + ", 成绩: " + pair._2());
        });
    }
}

运行结果：
在这里插入图片描述

sortByKey操作

说明

对 JavaPairRDD 数据按键进行排序的方法。默认为升序。
传入参数为Boolean值：true 升序，false降序

案例

将学生成绩按照成绩从高到低降序排序

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;
    @PostConstruct
    public void sortByKeyDemo(){
    // 定义原始数据
        List<Tuple2<String, Integer>> students = Arrays.asList(
                new Tuple2<>("张三", 85),
                new Tuple2<>("李四", 90),
                new Tuple2<>("王五", 65),
                new Tuple2<>("赵六", 95),
                new Tuple2<>("孙七", 90)
        );
// 生成RDD
        JavaPairRDD<String, Integer> studentsPairRDD = javaSparkContext.parallelizePairs(students);

        // 将学生原集合（姓名，成绩）格式的数据转换为（成绩，姓名）格式的键值对
        JavaPairRDD<Integer, String> integerStringJavaPairRDD = studentsPairRDD.mapToPair(item -> new Tuple2<>(item._2(), item._1()));

// 将（成绩，姓名）格式的键值对按照键降序排序，即按照成绩降序排序
        JavaPairRDD<Integer, String> stringIntegerJavaPairRDD = integerStringJavaPairRDD.sortByKey(false);

// 打印结果
        stringIntegerJavaPairRDD .collect() .forEach(pair -> {
            System.out.println("成绩: " + pair._1() + ", 姓名: " + pair._2());
        });
    }
}

运行结果：
在这里插入图片描述

行动算子操作

行动算子（Action）是指会触发 Spark 作业的执行，并且会产生一个结果或者副作用的操作。与转换算子（Transformation）不同，转换算子只会定义数据转换的计算逻辑，而不会立即执行。只有在遇到行动算子时，Spark 才会真正开始计算，并将结果返回给用户或写入外部存储。

当你调用一个行动算子时，Spark 会从头开始执行所有必要的转换操作，并将结果返回给你或者存储到外部系统（如 HDFS、数据库等）。

行动算子通常会返回一个具体的结果，例如一个列表、一个数值，或者在某些情况下，可能会执行一些副作用操作（例如将数据写入磁盘）。

collect 操作

说明

将分布式 RDD 中的所有数据项拉取到本地驱动程序（Driver）中，通常作为一个数组、列表或其他集合类型。因为 collect() 会将整个 RDD 数据集拉到本地，所以如果数据量非常大，可能会导致内存溢出（OutOfMemoryError）。

注意事项：
数据量大时的风险：如果 RDD 中包含的数据量非常大，调用 collect() 会导致所有数据被加载到本地驱动程序的内存中，这可能会导致内存溢出错误（OutOfMemoryError）。因此，建议在数据集非常大的情况下谨慎使用 collect()。

建议：对于大规模数据集，通常会使用其他行动算子（如 take()）获取一个数据的子集，避免一次性加载所有数据。

并行计算的代价：尽管 collect() 会将所有数据从分布式环境中拉回到单个节点，但它不会对数据进行额外的计算，只会执行之前定义的转换操作。因此，它本质上是将整个数据集的计算结果从集群中汇总回来。

案例

收集RDD数据并打印。

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;
    @PostConstruct
    public void collectDemo(){
        // 定义一个整数集合
        List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // collect()方法，用于将RDD中的数据收集到Driver端，并返回一个List。
        List<Integer> collect = parallelize.collect();

        // 打印结果
        collect.forEach(item -> System.out.println(item));
    }
}

执行结果：
在这里插入图片描述

count 操作

说明

统计RDD中的元素数量。

案例

统计RDD中的元素数量，并输出


import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;
    @PostConstruct
    public void countDemo(){
        // 定义一个整数集合
        List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // count()方法，统计数量。
        Long total = parallelize.count();

        // 打印结果
        System.out.println("元素数量 = " + total);
    }
}

运行结果：
在这里插入图片描述

first操作

说明

返回RDD中的第一个元素

案例

获取RDD中的第一个元素并打印

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void firstDemo(){
        // 创建一个集合
        List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // 获取第一个元素
        Integer first = parallelize.first();

        // 打印结果
        System.out.println("第一个元素 = " + first);
    }
}

运行结果：在这里插入图片描述

take操作

说明

在RDD中从头获取指定数量的元素，返回获取的元素集合。

案例

获取前3个元素

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void takeDemo(){
        // 创建一个集合
        List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);

        // 创建RDD
        JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);

        // 获取前三个元素
        List<Integer> takes = parallelize.take(3);

        // 打印结果
        System.out.println("前三个元素 = " + takes);
    }
}

运行结果：
在这里插入图片描述

countByKey操作

说明

统计RDD中每种键的数量。

案例

根据键值对，统计键值对中不同键的数量。并打印。

import lombok.extern.slf4j.Slf4j;
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.PairFunction;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    @PostConstruct
    public void countByKeyDemo(){
        // 创建一个集合
        List<Tuple2<String,Integer>> arrs = Arrays.asList(
                new Tuple2<String,Integer>("张三",15),
                new Tuple2<String,Integer>("张三",20),
                new Tuple2<String,Integer>("李四",20),
                new Tuple2<String,Integer>("李四",30),
                new Tuple2<String,Integer>("李四",50),
                new Tuple2<String,Integer>("王五",10)
        );

        // 创建JavaPairRDD 对象，JavaPairRDD对象的元素为键值对。
        JavaPairRDD<String, Integer> parallelize = javaSparkContext.parallelizePairs(arrs);

        // 使用 countByKey() 方法，统计相同键的数量。
        Map<String, Long> countByKey = parallelize.countByKey();

        // 打印结果
        countByKey.forEach((key, value) -> System.out.println("键: " + key + ", 数量: " + value));
    }
}

运行结果：
在这里插入图片描述

saveAsTextFile 操作

说明

将 RDD 的数据以txt文件保存到外部存储系统。传入指定路径，文件会生成到该路径下。

注意
输出路径不能存在：如果输出路径已经存在，saveAsTextFile 会抛出异常
文件分区：Spark 会将每个分区的数据写入一个独立的文件。因此，如果 RDD 有多个分区，它会生成多个文件，每个文件对应一个分区的数据 。文件名会根据分区编号进行自动命名，通常形式是：part-00***
文本格式：保存时，RDD 中的每个元素会被转换为文本行。默认情况下，Spark 会把 RDD 中的每个元素的 toString() 输出到文件中。
路径支持分布式存储：saveAsTextFile 支持将数据保存到本地文件系统、HDFS、S3 等分布式存储中。路径的格式取决于存储系统的类型。例如，如果要保存到 HDFS，路径应该以 hdfs:// 开头。

案例

将RDD存入外部文件。

以下案例中，将基于集合生成的RDD保存到saveRddDemo文件夹中。具体参考本文中的 RDD存入外部文件中 内容

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;

/**
 * spark 案例
 */
@Slf4j
@Component
public class DemoTimer {
    @Autowired
    JavaSparkContext javaSparkContext;

    @Autowired
    SparkSession sparkSession;

    /**
     * 创建 RDD 算子
     */
    @PostConstruct
    public void createRDD() {
        // 从集合创建 RDD
        List<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");
        
        // parallelize(): 创建RDD，RDD中创建默认分区数
        // parallelize( 元素, 分区数)： 创建RDD，RDD中指定分区数
        JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDD
        
        parallelize.collect().forEach(System.out::println); // 打印

        // 存储的目录文件夹路径。此处为项目中的路径且目录必须为不存在的路径。
        String fileSavePath="I:\\zhang\\SpringBootSparkDemo\\src\\main\\resources\\saveRddDemo";
        parallelize.saveAsTextFile(fileSavePath);// 开始将RDD保存到文件中 
    }
}

foreach操作

说明

循环遍历RDD中的元素

案例

以上案例中大部分用到此方法，是否用方式看以上案例。

原文地址：https://blog.csdn.net/qq_38313548/article/details/144027674

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Nodemailer使用教程：在Node.js中发送电子邮件
下一篇：️ 爬虫开发中常见的性能优化策略有哪些？

基于spring boot的医院挂号就诊系统论文
第二点：设计有效的导航，这个包括每个页面上都有导航条的显示，有时也可以在页面的底部设计导航条，当用户进入具体页面时，要设计相应的位置提示，在页面中比较特殊的位置，需要设计返回链接，可以返回上个页面，也
阅读更多2024-11-28
【Unity踩坑】Unity中父对象是非均匀缩放时出现倾斜或剪切现象
本文解决了Unity中父对象是非均匀缩放时出现倾斜或剪切的问题
阅读更多2024-11-28
「Mac玩转仓颉内测版29」基础篇9 - 数组类型详解
本篇将介绍 Cangjie 中的数组类型，包括数组的定义、创建、访问、常见操作以及在数据管理中的应用，帮助开发者熟练掌握数组的使用。
阅读更多2024-11-28
创建mapreduce项目使用maven
创建mapreduce项目使用maven
阅读更多2024-11-28
Python开源项目周排行 2024年第16周
注:当前文章会不定期进行更新。标签： ERP PDF OCR。
阅读更多2024-11-28
使用EFK收集k8s日志
首先我们使用EFK收集Kubernetes集群中的日志，本次实验讲解的是在Kubernetes集群中启动一个Elasticsearch集群，如果企业内已经有了Elasticsearch集群，可以直接将
阅读更多2024-11-28
C#基础41-45
求n以内（不包括n）同时能被3和7整数的所有自然数之和的平方根s，并作为函数值返回，最后结果s输出到控制台。44.数组x中有n个数，求出奇数的个数cn1和偶数的个数cn2以及数组x下标为奇数的元素值的
阅读更多2024-11-28
JavaScript：遍历一个对象数组，获取每个对象中 mainKeyId 对应的值，并以这个值作为新属性的键名，给该对象添加一个新属性。
假设我们有一个对象数组，每个对象包含一个mainKeyId字段，以mainKeyId的值作为新的键名，并给它添加一个新的属性。
阅读更多2024-11-28
【AI系统】谷歌 TPU v4 与光路交换
TPU v4 通过采用 7nm 工艺，翻倍的 MXU 数量，以及显著增加的内存容量和带宽，实现了性能的大幅提升。特别引人注目的是 Sparse Core 的引入，它专门针对稀疏计算进行了优化，使得 T
阅读更多2024-11-28
SpringBoot文件上传之秒传、断点续传、分片上传
比如，在线教育平台上的视频资源上传，社交平台上的图片分享，以及企业内部的知识文档管理等。这些场景对文件上传的要求也各不相同，有的追求速度，有的注重稳定性，还有的需要考虑文件大小和安全性。这对于大文件上
阅读更多2024-11-28

【spark-spring boot】学习笔记

目录

说明

RDD学习

RDD介绍

RDD案例

基于集合创建RDD

RDD存入外部文件中

转换算子 操作

map 操作

说明

案例

flatMap操作

说明

案例

filter 操作

说明

案例

groupBy 操作

说明

案例

distinct 操作

说明

案例

sortBy 操作

说明

案例

mapToPair 操作

说明

案例

mapValues操作

说明

案例

groupByKey操作

说明

案例

reduceByKey操作

说明

案例

sortByKey操作

说明

案例

行动算子 操作

collect 操作

说明

案例

count 操作

说明

案例

first操作

说明

案例

take操作

说明

案例

countByKey操作

说明

案例

saveAsTextFile 操作

说明

案例

foreach操作

说明

案例

相关文章

转换算子操作

行动算子操作