Spring boot项目实战——spark环境搭建及RDD运算

环境搭建

自己查资料了解spark的部署环境。本项目是本地环境，伪分布式。

在window上使用spark必须现在本地搭建hadoop环境，具体请参考之前的文章windows上配置hadoop并通过idea连接本地spark和服务器spark

Spark环境搭建完成后，正常创建spring boot程序并启动生产bean：

//生产bean
@Bean(name = "sc")
public JavaSparkContext makeSparkContext(){
    
    return new JavaSparkContext("local", "spark-container");
}

SparkContext是rdd和context，在java中是JavaSparkContext，所有的操作都是基于这个对象。

最好尽早配置程序：

System.setProperty("hadoop.home.dir","D:\\SoftWares\\Apache\\spark-3.3.1-bin-hadoop3");防止找不到spark。

JavaSparkContext的使用

//doubles为一个浮点型数组

//内存中加载数据
JavaRDD<Double> rdd = sc.parallelize(doubles);

Double reduce = rdd.reduce((Function2<Double, Double, Double>) (aDouble, aDouble2) -> aDouble + aDouble2);

通过JavaSparkContext记录到内存中，通过reduce计算，并返回计算结果。

代码的核心部分在于生产第一个RDD，通过parallelize方法。其他方法请参考Spark核心RDD详解（设计和运行原理、分区、创建、转换、动作和持久化）

构建 RDD 后，可以通过 RDD 操作和转换来操作数据。

@RestController
@RequestMapping("/analyze")
public class SaleTotalController implements Serializable {
    

    @Resource
    private JavaSparkContext sc;

    //
    @GetMapping("/consumption")
    public SaleJson getTotal(){
    
        ReadCSV readCSV = new ReadCSV();
        List<String> columnList = readCSV.getColumnList("C:\\Users\\fireapproval\\Desktop\\数据集\\test.csv", 7);
        columnList.remove(0);

        //System.out.println(columnList);
        List<Double> doubles = new ArrayList<>();
        for (String str:columnList
             ) {
    
            doubles.add(Double.parseDouble(str));
        }
        //System.out.println(doubles);

        //System.out.println(doubles.get(2).getClass());

        //内存中加载数据
        JavaRDD<Double> rdd = sc.parallelize(doubles);

        Double reduce = rdd.reduce((Function2<Double, Double, Double>) (aDouble, aDouble2) -> aDouble + aDouble2);


        //System.out.println(reduce);

        SaleJson saleJson = new SaleJson();
        saleJson.setTotal(reduce);
        return saleJson;
    }
}

上述计算的逻辑代码应该放在service层，这里为了演示目的。

以下是词频统计的核心RDD操作：

JavaRDD<String> rdd = sc.parallelize(tmpList);

//JavaRDD<String> rdd1 = rdd.map(x -> x.concat("xxx"));

//转化为(word,1)类型
JavaPairRDD<String, Integer> wordMap = rdd.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));

//对(word,1)计算 //必须要用匿名函数

JavaPairRDD<String, Integer> reduceMap = wordMap.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);


//map接收rdd
Map<String, Integer> rddMap = reduceMap.collectAsMap();