SpringBoot使用Spark过程详解

 更新时间:2023年02月15日 10:46:31   作者:我有一只肥螳螂  
这篇文章主要介绍SpringBoot使用Spark的方法的相关知识,小编通过实际案例向大家展示操作过程,操作方法简单快捷,实用性强,希望能帮助大家解决问题

前提: 可以参考文章 SpringBoot 接入 Spark

  • SpringBoot 已经接入 Spark
  • 已配置 JavaSparkContext
  • 已配置 SparkSession
@Resource
private SparkSession sparkSession;
@Resource
private JavaSparkContext javaSparkContext;

读取 txt 文件

测试文件 word.txt

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 个

读取 csv 文件

测试文件 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());
    //输出结果
    System.out.println(wordsRDD.collect());
}

输出结果

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

结果输出

读取 Json 文件

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

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

注意:testjson.json 文件的内容不能带格式,需要进行压缩

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

输出结果

中文输出乱码

测试文件 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());
    //输出结果
    System.out.println(wordsRDD.collect());
}

输出结果,发现中文乱码,可恶

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

输出结果

到此这篇关于SpringBoot使用Spark过程详解的文章就介绍到这了,更多相关SpringBoot Spark内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:

相关文章

  • SpringBoot快速整合Mybatis、MybatisPlus(代码生成器)实现数据库访问功能

    SpringBoot快速整合Mybatis、MybatisPlus(代码生成器)实现数据库访问功能

    这篇文章主要介绍了SpringBoot快速整合Mybatis、MybatisPlus(代码生成器)实现数据库访问功能,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-04-04
  • Spring实现声明式事务的方法详解

    Spring实现声明式事务的方法详解

    这篇文章主要介绍了Spring实现声明式事务的方法详解,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-01-01
  • 解决idea中svn提交时performing vcs refresh时间很长的问题

    解决idea中svn提交时performing vcs refresh时间很长的问题

    这篇文章主要介绍了解决idea中svn提交时performing vcs refresh时间很长的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-09-09
  • Spring中的策略模式简单实现与使用分析

    Spring中的策略模式简单实现与使用分析

    这篇文章主要介绍了Spring中的策略模式简单实现与使用分析,去初始化时除了 initMultipartResolver(上传文件)没有获取 Properties defaultStrategies;默认策略,其他的八大件都会使用到策略模式,需要的朋友可以参考下
    2024-01-01
  • SpringBoot使用Thymeleaf自定义标签的实例代码

    SpringBoot使用Thymeleaf自定义标签的实例代码

    这篇文章主要介绍了SpringBoot使用Thymeleaf自定义标签的实例代码,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-09-09
  • java equals和=,==的区别详细介绍

    java equals和=,==的区别详细介绍

    这篇文章主要介绍了java equals和=,==的区别,学习Java的朋友对equals 和== 这个概念开始使用的时候会有疑问,很难辨别如何正确使用,这里帮大家详细讲解该知识点,希望大家能掌握,有需要的小伙伴可以参考下
    2016-10-10
  • 详细分析JVM类加载机制

    详细分析JVM类加载机制

    JVM将class文件字节码文件加载到内存中, 并将这些静态数据转换成方法区中的运行时数据结构,在堆(并不一定在堆中,HotSpot在方法区中)中生成一个代表这个类的java.lang.Class 对象,作为方法区类数据的访问入口,接下来将详细讲解JVM类加载机制
    2022-04-04
  • Java消息队列RabbitMQ之消息回调详解

    Java消息队列RabbitMQ之消息回调详解

    这篇文章主要介绍了Java消息队列RabbitMQ之消息回调详解,消息回调,其实就是消息确认(生产者推送消息成功,消费者接收消息成功)  , 对于程序来说,发送者没法确认是否发送成功,需要的朋友可以参考下
    2023-07-07
  • Java中如何避免sql注入实例详解

    Java中如何避免sql注入实例详解

    SQL注入是最常见的攻击方式之一,它不是利用操作系统或其它系统的漏洞来实现攻击的,而是程序员因为没有做好判断,被不法用户钻了SQL的空子,下面这篇文章主要给大家介绍了关于Java中如何避免sql注入的相关资料,需要的朋友可以参考下
    2022-01-01
  • xxl-job对比ElasticJob使用示例详解

    xxl-job对比ElasticJob使用示例详解

    这篇文章主要为大家介绍了xxl-job对比ElasticJob使用示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2022-06-06

最新评论