深入了解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数据加载与保存内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

详解JAVA抓取网页的图片,JAVA利用正则表达式抓取网站图片
这篇文章主要介绍了详解JAVA抓取网页的图片,JAVA利用正则表达式抓取网站图片,非常具有实用价值，需要的朋友可以参考下。
2016-12-12
Java中如何模拟HTTP请求并验证功能
要模拟HTTP请求并验证功能,你可以使用Spring Boot提供的MockMvc工具,它允许我们在没有实际启动HTTP服务器的情况下测试Spring MVC控制器,下面给大家分享如何模拟HTTP请求并验证功能,感兴趣的朋友一起看看吧
2024-05-05
JavaEE SpringMyBatis是什么? 它和Hibernate的区别及如何配置MyBatis
这篇文章主要介绍了JavaEE Spring MyBatis是什么? 它和Hibernate的区别有哪些？如何配置MyBatis?本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-08-08
Java订单30分钟未支付自动取消该怎么实现
在开发中往往会遇到一些关于延时任务的需求,例如生成订单30分钟未支付,则自动取消,下面这篇文章主要给大家介绍了关于Java订单30分钟未支付自动取消该怎么实现的相关资料,需要的朋友可以参考下
2023-03-03
Java实现读取Jar文件属性的方法详解
这篇文章主要为大家详细介绍了如何利用Java语言实现读取Jar文件属性的功能，文中的示例代码讲解详细，感兴趣的小伙伴可以跟随小编一起学习一下
2022-08-08
Mybatis反射核心类Reflector的实现
本文主要介绍了Mybatis反射核心类Reflector的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-11-11
Java实现下载文件的6种方式
本文主要介绍了Java实现下载文件的6种方式，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-06-06
Java中面向对象的知识点总结
Java是一门面向对象的语言。对象是Java程序中的基本实体。除了对象之外Java程序同样处理基本数据。下面这篇文章主要给大家总结了关于Java中面向对象的知识点，需要的朋友可以参考借鉴，下面来一起看看吧。
2017-02-02
Java Collection 接口和常用方法综合详解
Collection派生出三个子接口，Set代表不可重复的无序集合、List代表可重复的有序集合、Queue是java提供的队列实现，通过它们不断的扩展出很多的集合类，接下来我们详细介绍一下
2021-11-11
MyBatis 动态SQL使用及原理
这篇文章主要为大家介绍了MyBatis动态SQL使用及原理的学习详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-05-05