深入了解SparkSQL中数据的加载与保存
更新时间：2023年11月15日 08:37:20 作者：shangjg3
这篇文章主要为大家详细介绍了SparkSQL中数据的加载与保存的相关知识,文中的示例代码讲解详细,具有一定的学习价值,感兴趣的小伙伴可以了解下
1 读取和保存文件
SparkSQL读取和保存的文件一般为三种，JSON文件、CSV文件和列式存储的文件，同时可以通过添加参数，来识别不同的存储和压缩格式。
1.1 CSV文件
1）代码实现
package com.atguigu.sparksql;
import com.atguigu.sparksql.Bean.User;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;
public class Test06_CSV {
public static void main(String[] args) throws ClassNotFoundException {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
DataFrameReader reader = spark.read();
// 添加参数 读取csv
Dataset<Row> userDS = reader
.option("header", "true")//默认为false 不读取列名
.option("sep",",") // 默认为, 列的分割
// 不需要写压缩格式 自适应
.csv("input/user.csv");
userDS.show();
// 转换为user的ds
// 直接转换类型会报错 csv读取的数据都是string
// Dataset<User> userDS1 = userDS.as(Encoders.bean(User.class));
userDS.printSchema();
Dataset<User> userDS1 = userDS.map(new MapFunction<Row, User>() {
@Override
public User call(Row value) throws Exception {
return new User(Long.valueOf(value.getString(0)), value.getString(1));
}
}, Encoders.bean(User.class));
userDS1.show();
// 写出为csv文件
DataFrameWriter<User> writer = userDS1.write();
writer.option("header",";")
.option("header","true")
// .option("compression","gzip")// 压缩格式
// 写出模式
// append 追加
// Ignore 忽略本次写出
// Overwrite 覆盖写
// ErrorIfExists 如果存在报错
.mode(SaveMode.Append)
.csv("output");
//4. 关闭sparkSession
spark.close();
}
}1.2 JSON文件
package com.atguigu.sparksql;
import com.atguigu.sparksql.Bean.User;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
public class Test07_JSON {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
Dataset<Row> json = spark.read().json("input/user.json");
// json数据可以读取数据的数据类型
Dataset<User> userDS = json.as(Encoders.bean(User.class));
userDS.show();
// 读取别的类型的数据也能写出为json
DataFrameWriter<User> writer = userDS.write();
writer.json("output1");
//4. 关闭sparkSession
spark.close();
}
}1.3 Parquet文件
列式存储的数据自带列分割。
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class Test08_Parquet {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
Dataset<Row> json = spark.read().json("input/user.json");
// 写出默认使用snappy压缩
// json.write().parquet("output");
// 读取parquet 自带解析 能够识别列名
Dataset<Row> parquet = spark.read().parquet("output");
parquet.printSchema();
//4. 关闭sparkSession
spark.close();
}
}2 与MySQL交互
1）导入依赖
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>2）从MySQL读数据
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import java.util.Properties;
public class Test09_Table {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
Dataset<Row> json = spark.read().json("input/user.json");
// 添加参数
Properties properties = new Properties();
properties.setProperty("user","root");
properties.setProperty("password","000000");
// json.write()
// // 写出模式针对于表格追加覆盖
// .mode(SaveMode.Append)
// .jdbc("jdbc:mysql://hadoop102:3306","gmall.testInfo",properties);
Dataset<Row> jdbc = spark.read().jdbc("jdbc:mysql://hadoop102:3306", "gmall.testInfo", properties);
jdbc.show();
//4. 关闭sparkSession
spark.close();
}
}3 与Hive交互
SparkSQL可以采用内嵌Hive（spark开箱即用的hive），也可以采用外部Hive。企业开发中，通常采用外部Hive。
3.1 Linux中的交互
1）添加MySQL连接驱动到spark-yarn的jars目录
[atguigu@hadoop102 spark-yarn]$ cp /opt/software/mysql-connector-java-5.1.27-bin.jar /opt/module/spark-yarn/jars
2）添加hive-site.xml文件到spark-yarn的conf目录
[atguigu@hadoop102 spark-yarn]$ cp /opt/module/hive/conf/hive-site.xml /opt/module/spark-yarn/conf
3）启动spark-sql的客户端即可
[atguigu@hadoop102 spark-yarn]$ bin/spark-sql --master yarn
spark-sql (default)> show tables;
3.2 IDEA中的交互
1）添加依赖
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.1</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.1</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
</dependency>
</dependencies>2）拷贝hive-site.xml到resources目录（如果需要操作Hadoop，需要拷贝hdfs-site.xml、core-site.xml、yarn-site.xml）
3）代码实现
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
public class Test10_Hive {
public static void main(String[] args) {
System.setProperty("HADOOP_USER_NAME","atguigu");
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder()
.enableHiveSupport()// 添加hive支持
.config(conf).getOrCreate();
//3. 编写代码
spark.sql("show tables").show();
spark.sql("create table user_info(name String,age bigint)");
spark.sql("insert into table user_info values('zhangsan',10)");
spark.sql("select * from user_info").show();
//4. 关闭sparkSession
spark.close();
}
}到此这篇关于深入了解SparkSQL中数据的加载与保存的文章就介绍到这了,更多相关SparkSQL数据加载与保存内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了IntelliJ IDEA 2020 的Debug功能，本文通过实例截图相结合给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-04-04
这篇文章主要介绍了SpringMVC拦截器实现单点登录，简单介绍了springmvc拦截器，单点登录实现原理等相关内容，具有一定参考价值，需要的朋友可以了解下。2017-11-11
Java try-catch-finally异常处理机制详解
这篇文章主要介绍了Java try-catch-finally异常处理机制详解,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下2021-08-08
这篇文章主要介绍了Spring中的AOP原理与使用详解,AOP意为面向切面编程,可以通过预编译方式或运行期动态代理实现在不修改源代码的情况下给程序动态统一添加功能的一种技术,需要的朋友可以参考下2023-12-12
Spring Boot Admin的使用详解(Actuator监控接口)
这篇文章主要介绍了Spring Boot Admin的使用详解(Actuator监控接口)，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2021-05-05
这篇文章主要介绍了java 数据库连接与增删改查操作,结合实例形式详细分析了java使用jdbc进行数据库连接及增删改查等相关操作实现技巧与注意事项,需要的朋友可以参考下2019-11-11
在公司项目中,如果需要做一些公共的功能,如日志等,最好的方式是使用自定义注解,自定义注解可以实现我们对想要添加日志的方法上添加,这篇文章基于日志功能来讲讲自定义注解应该如何开发和使用,需要的朋友可以参考下2023-08-08
这篇文章主要为大家分享了java实现163邮箱发送邮件到qq邮箱成功案例，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2016-05-05
作者是一个来自河源的大三在校生，以下笔记都是作者自学之路的一些浅薄经验，如有错误请指正，将来会不断的完善笔记，帮助更多的Java爱好者入门2022-03-03
本文主要介绍Struts2的路径问题，尽量不要使用相对路径，使用相对路径会让路径问题变得很繁琐很麻烦，推荐使用绝对路径，希望能给大家做一个参考。2016-06-06
最新评论