SpringBoot使用Spark過程詳解

更新時(shí)間：2023年02月15日 10:46:31 作者：我有一只肥螳螂

這篇文章主要介紹SpringBoot使用Spark的方法的相關(guān)知識(shí)，小編通過實(shí)際案例向大家展示操作過程，操作方法簡(jiǎn)單快捷，實(shí)用性強(qiáng)，希望能幫助大家解決問題

讀取 txt 文件

測(cè)試文件 word.txt

java 代碼

textFile：獲取文件內(nèi)容，返回 JavaRDD
flatMap：過濾數(shù)據(jù)
mapToPair：把每個(gè)元素都轉(zhuǎn)換成一個(gè)<K,V>類型的對(duì)象，如 <123,1>,<456,1>
reduceByKey：對(duì)相同key的數(shù)據(jù)集進(jìn)行預(yù)聚合

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);
    //輸出結(jié)果
    List<Tuple2<String, Integer>> result = wordAndCountRDD.collect();
    result.forEach(System.out::println);
}

結(jié)果得出，123 有 3 個(gè)，456 有 2 個(gè)，789 有 1 個(gè)

讀取 csv 文件

測(cè)試文件 testcsv.csv

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());
    //輸出結(jié)果
    System.out.println(wordsRDD.collect());
}

輸出結(jié)果

讀取 MySQL 數(shù)據(jù)庫(kù)表

format：獲取數(shù)據(jù)庫(kù)建議是 jdbc
option.url：添加 MySQL 連接 url
option.user：MySQL 用戶名
option.password：MySQL 用戶密碼
option.dbtable：sql 語(yǔ)句
option.driver：數(shù)據(jù)庫(kù) 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();
    //轉(zhuǎn)化為RDD
    JavaRDD<Row> rowJavaRDD = jdbcDF.javaRDD();
    System.out.println(rowJavaRDD.collect());
}

也可以把表內(nèi)容輸出到文件，添加以下代碼

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();

結(jié)果輸出

讀取 Json 文件

測(cè)試文件 testjson.json，內(nèi)容如下

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

注意：testjson.json 文件的內(nèi)容不能帶格式，需要進(jìn)行壓縮

java 代碼

createOrReplaceTempView：讀取 json 數(shù)據(jù)后，創(chuàng)建數(shù)據(jù)表 t
sparkSession.sql：使用 sql 對(duì) t 進(jìn)行查詢，輸出 age 大于 3 的數(shù)據(jù)

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());
}

輸出結(jié)果

中文輸出亂碼

測(cè)試文件 testcsv.csv

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());
    //輸出結(jié)果
    System.out.println(wordsRDD.collect());
}

輸出結(jié)果，發(fā)現(xiàn)中文亂碼，可惡

原因：textFile 讀取文件沒有解決亂碼問題，但 sparkSession.read() 卻不會(huì)亂碼

解決辦法：獲取文件方式由 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());
        //輸出結(jié)果
        System.out.println(gbkWordsRDD.collect());
    }

輸出結(jié)果