SpringBoot 使用 Spark

秒速五厘米 2024-03-27 16:29 56阅读 0赞

#### 文章目录 ####

*  读取 txt 文件
 *  读取 csv 文件
 *  读取 MySQL 数据库表
 *  读取 Json 文件
 *  中文输出乱码

前提： 可以参考文章 [SpringBoot 接入 Spark  
][SpringBoot _ Spark]

1.  SpringBoot 已经接入 Spark
2.  已配置 JavaSparkContext
3.  已配置 SparkSession

@Resource
    private SparkSession sparkSession;
    
    @Resource
    private JavaSparkContext javaSparkContext;

## 读取 txt 文件 ##

测试文件 word.txt  
![在这里插入图片描述][52b484ca1b91418e97ebc7bc6bcb5506.png]

java 代码

*  textFile：获取文件内容，返回 JavaRDD
 *  flatMap：过滤数据
 *  mapToPair：把每个元素都转换成一个<K,V>类型的对象，如 <123,1>,<456,1>
 *  reduceByKey：对相同key的数据集进行预聚合

public void testSparkText() {
        
        String file = "D:\\TEMP\\word.txt";
        JavaRDD<String> fileRDD =  javaSparkContext.textFile(file);
    
        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
        JavaPairRDD<String, Integer> wordAndOneRDD = wordsRDD.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> wordAndCountRDD = wordAndOneRDD.reduceByKey((a, b) -> a + b);
    
        //输出结果
        List<Tuple2<String, Integer>> result = wordAndCountRDD.collect();
        result.forEach(System.out::println);
    }

结果得出，123 有 3 个，456 有 2 个，789 有 1 个  
![在这里插入图片描述][b4651da3d9474880b8654f6481853858.png]

## 读取 csv 文件 ##

测试文件 testcsv.csv  
![在这里插入图片描述][d886e0545e754769b16a72aeb8a2545d.png]

java 代码

public void testSparkCsv() {
        
        String file = "D:\\TEMP\\testcsv.csv";
        JavaRDD<String> fileRDD = javaSparkContext.textFile(file);
        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());
    
        //输出结果
        System.out.println(wordsRDD.collect());
    }

输出结果  
![在这里插入图片描述][9ba9b3831848471cbdc27c18a7b16e87.png]

## 读取 MySQL 数据库表 ##

*  format：获取数据库建议是 jdbc
 *  option.url：添加 MySQL 连接 url
 *  option.user：MySQL 用户名
 *  option.password：MySQL 用户密码
 *  option.dbtable：sql 语句
 *  option.driver：数据库 driver，MySQL 使用 com.mysql.cj.jdbc.Driver
 *

public void testSparkMysql() throws IOException {
        
        
        Dataset<Row> jdbcDF = sparkSession.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://192.168.140.1:3306/user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai")
                .option("dbtable", "(SELECT * FROM xxxtable) tmp")
                .option("user", "root")
                .option("password", "xxxxxxxxxx*k")
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .load();
    
        jdbcDF.printSchema();
        jdbcDF.show();
    
        //转化为RDD
        JavaRDD<Row> rowJavaRDD = jdbcDF.javaRDD();
        System.out.println(rowJavaRDD.collect());
    }

也可以把表内容输出到文件，添加以下代码

List<Row> list = rowJavaRDD.collect();
    BufferedWriter bw;
    bw = new BufferedWriter(new FileWriter("d:/test.txt"));
    for (int j = 0; j < list.size(); j++) {
        
        bw.write(list.get(j).toString());
        bw.newLine();
        bw.flush();
    }
    bw.close();

结果输出  
![在这里插入图片描述][0567109d4c894aeb83899642e6358450.png]

## 读取 Json 文件 ##

测试文件 testjson.json，内容如下

[{
        
    	"name": "name1",
    	"age": "1"
    }, {
        
    	"name": "name2",
    	"age": "2"
    }, {
        
    	"name": "name3",
    	"age": "3"
    }, {
        
    	"name": "name4",
    	"age": "4"
    }]

注意：testjson.json 文件的内容不能带格式，需要进行压缩  
![在这里插入图片描述][d997a907d3e144b29baceb42285812b3.png]

java 代码

*  createOrReplaceTempView：读取 json 数据后，创建数据表 t
 *  sparkSession.sql：使用 sql 对 t 进行查询，输出 age 大于 3 的数据

public void testSparkJson() {
        
        Dataset<Row> df = sparkSession.read().json("D:\\TEMP\\testjson.json");
        df.printSchema();
    
        df.createOrReplaceTempView("t");
        Dataset<Row> row = sparkSession.sql("select age,name from t where age > 3");
    
        JavaRDD<Row> rowJavaRDD = row.javaRDD();
        System.out.println(rowJavaRDD.collect());
    }

输出结果  
![在这里插入图片描述][cd6abd09007740528fc81add88b7c6c7.png]

## 中文输出乱码 ##

测试文件 testcsv.csv  
![在这里插入图片描述][38d1b24af3cd433191bf5718c7fae785.png]

输出结果，发现中文乱码，可恶  
![在这里插入图片描述][471eab29525e4bccbd7184b51223ca9e.png]  
原因：textFile 读取文件没有解决乱码问题，但 sparkSession.read() 却不会乱码  
解决办法：获取文件方式由 textFile 改成 hadoopFile，由 hadoopFile 指定具体编码

public void testSparkCsv() {
        
            String file = "D:\\TEMP\\testcsv.csv";
            String code = "gbk";
            
            JavaRDD<String> gbkRDD = javaSparkContext.hadoopFile(file, TextInputFormat.class, LongWritable.class, Text.class).map(p -> new String(p._2.getBytes(), 0, p._2.getLength(), code));
            JavaRDD<String> gbkWordsRDD = gbkRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());
    
            //输出结果
            System.out.println(gbkWordsRDD.collect());
        }

输出结果  
![在这里插入图片描述][b1474d8a0d2a4db48e4ae312ae7610e8.png]

[SpringBoot _ Spark]: https://blog.csdn.net/weixin_42555971/article/details/128974534
[52b484ca1b91418e97ebc7bc6bcb5506.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/7bb509cc42044d50b8947846700a9e2d.png
[b4651da3d9474880b8654f6481853858.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/d2a405b27a6a428aac9dd5144e94f061.png
[d886e0545e754769b16a72aeb8a2545d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/487ccfaa1a3944ed89b3684275de38dc.png
[9ba9b3831848471cbdc27c18a7b16e87.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/d0ad4186e4cc49ddacda8655f106ac6f.png
[0567109d4c894aeb83899642e6358450.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/4f2381f95e5649c79fab16968f75c250.png
[d997a907d3e144b29baceb42285812b3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/1874105567864e5fa48b32a200bb771e.png
[cd6abd09007740528fc81add88b7c6c7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/a5949fe778f341f191c39eac3ac30840.png
[38d1b24af3cd433191bf5718c7fae785.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/b711aaa8db2d430cb1f01ca1eca244db.png
[471eab29525e4bccbd7184b51223ca9e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/68b93ba4b4a24700b1828981a3ab589b.png
[b1474d8a0d2a4db48e4ae312ae7610e8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/6a7ff02a5ed24011ad2f21fe4c068190.png