SpringBoot 集成 Spark简单实现例子

旧城等待， 2023-10-01 00:21 22阅读 0赞

### 一：配置文件pom.xml ###

<dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.13</artifactId>
                <version>3.2.0</version>
            </dependency>
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>6.0.6</version>
            </dependency>
            <dependency>
                <groupId>org.codehaus.janino</groupId>
                <artifactId>janino</artifactId>
                <version>3.0.8</version>
            </dependency>

### 二：测试类 ###

package com.zkaw.hadoop.spark;
    
    import org.apache.spark.SparkConf;
    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.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    import scala.Tuple2;
    
    import java.util.Arrays;
    import java.util.List;
    
    
    /** @Author: best_liu
     * @Description: SpringBoot 集成 Spark demo
     * @Date: 13:58 2023/4/24
     * @Param
     * @return
     **/
    public class SparkTest {
    
    
        public static void main(String[] args) {
    //        testSparkRddTxt();
    
    //        testSparkRddCsv();
    
            testSparkRddMysql();
    
    //        testSparkRddJson();
        }
    
    
        //spark 测试外部文件（txt）
        //rdd:resilient distributed dataset ，弹性分布式数据集
        public static void testSparkRddTxt(){
            //1.环境准备
            SparkConf sparkConf = new SparkConf();
            sparkConf.set("spark.driver.host","localhost");
            //sparkConf.set("SPARK_LOCAL_HOSTNAME","localhost");
            sparkConf.setAppName("JavaSparkDemo").setMaster("local[*]");
    
            JavaSparkContext jsc = new JavaSparkContext(sparkConf);
            jsc.setLogLevel("WARN");
    
            //2.处理数据
            JavaRDD<String> fileRDD = jsc.textFile("D:\\test.txt");
    
            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);
    
            //3.输出结果
            List<Tuple2<String, Integer>> result = wordAndCountRDD.collect();
            result.forEach(System.out::println);
    
            //4.关闭资源
            jsc.stop();
        }
    
        //spark 测试外部文件（csv）
        public static void testSparkRddCsv(){
            //1.环境准备
            SparkConf sparkConf = new SparkConf();
            sparkConf.set("spark.driver.host","localhost");
            sparkConf.setAppName("JavaSparkDemo").setMaster("local[*]");
    
            JavaSparkContext jsc = new JavaSparkContext(sparkConf);
            jsc.setLogLevel("WARN");
    
            //2.读取外部文件创建rdd,以字符串读取
            JavaRDD<String> fileRDD = jsc.textFile("D:\\testcsv.csv");
            //3.把文件内容使用,分割
            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);
    
            System.out.println(wordAndCountRDD.collect());
            jsc.stop();
    
        }
    
    
        //spark 操作mysql数据库
        //添加依赖：mysql-connector-java
        public static void testSparkRddMysql(){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest3")
                    .config("spark.driver.host", "localhost")
                    .config("spark.some.config.option", "some-value")
                    .master("local[*]")
                    .getOrCreate();
    
    
            //DataSet 是具有强类型的数据集合
            Dataset<Row> jdbcDF = spark.read()
                    .format("jdbc")
                    .option("url", "jdbc:mysql://127.0.0.1:3306/mybatis_plus?useSSL=false&useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8")
                    .option("dbtable", "(SELECT * FROM movie) tmp")
                    .option("user", "root")
                    .option("password", "root3306")
                    .option("driver","com.mysql.cj.jdbc.Driver")
                    .load();
    
            jdbcDF.printSchema();
            jdbcDF.show();
    
            //转化为RDD
            JavaRDD<Row> rowJavaRDD = jdbcDF.javaRDD();
            System.out.println(rowJavaRDD.collect());
    
            spark.stop();
        }
    
    
        //spark 测试json
        public static void testSparkRddJson(){
            //1.环境准备
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest3")
                    .config("spark.driver.host", "localhost")
                    .config("spark.some.config.option", "some-value")
                    .master("local[*]")
                    .getOrCreate();
    
            Dataset<Row> df = spark.read().json("D:\\testjson.json");
            df.printSchema();
            df.show();
    
            df.createOrReplaceTempView("t_person");
            spark.sql("select dictValue,dictLabel from t_person").show();
    
            spark.stop();
        }
    
    }

### 三：总结 ###

测试类中一共有写了四个测试方法，包含分析txt文件，csv文件，json数据处理，直连my[sql数据库][sql]，方法都经过测试，可以正常打印结果，特别是jdbc 直连mysql可以直接写sql语句，很方便

[sql]: https://so.csdn.net/so/search?q=sql%E6%95%B0%E6%8D%AE%E5%BA%93&spm=1001.2101.3001.7020