深入了解SparkSQL中數(shù)據(jù)的加載與保存

更新時(shí)間：2023年11月15日 08:37:20 作者：shangjg3

這篇文章主要為大家詳細(xì)介紹了SparkSQL中數(shù)據(jù)的加載與保存的相關(guān)知識(shí),文中的示例代碼講解詳細(xì),具有一定的學(xué)習(xí)價(jià)值,感興趣的小伙伴可以了解下

1 讀取和保存文件

SparkSQL讀取和保存的文件一般為三種，JSON文件、CSV文件和列式存儲(chǔ)的文件，同時(shí)可以通過(guò)添加參數(shù)，來(lái)識(shí)別不同的存儲(chǔ)和壓縮格式。

1.1 CSV文件

1）代碼實(shí)現(xiàn)

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. 創(chuàng)建配置對(duì)象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
        //2. 獲取sparkSession
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        //3. 編寫(xiě)代碼
        DataFrameReader reader = spark.read();
        // 添加參數(shù)  讀取csv
        Dataset<Row> userDS = reader
                .option("header", "true")//默認(rèn)為false 不讀取列名
                .option("sep",",") // 默認(rèn)為, 列的分割
                // 不需要寫(xiě)壓縮格式  自適應(yīng)
                .csv("input/user.csv");
        userDS.show();
        // 轉(zhuǎn)換為user的ds
        // 直接轉(zhuǎn)換類型會(huì)報(bào)錯(cuò)  csv讀取的數(shù)據(jù)都是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();
        // 寫(xiě)出為csv文件
        DataFrameWriter<User> writer = userDS1.write();
        writer.option("header",";")
                .option("header","true")
//                .option("compression","gzip")// 壓縮格式
                // 寫(xiě)出模式
                // append 追加
                // Ignore 忽略本次寫(xiě)出
                // Overwrite 覆蓋寫(xiě)
                // ErrorIfExists 如果存在報(bào)錯(cuò)
                .mode(SaveMode.Append)
                .csv("output");
        //4. 關(guān)閉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. 創(chuàng)建配置對(duì)象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");

        //2. 獲取sparkSession
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        //3. 編寫(xiě)代碼
        Dataset<Row> json = spark.read().json("input/user.json");

        // json數(shù)據(jù)可以讀取數(shù)據(jù)的數(shù)據(jù)類型
        Dataset<User> userDS = json.as(Encoders.bean(User.class));
        userDS.show();

        // 讀取別的類型的數(shù)據(jù)也能寫(xiě)出為json
        DataFrameWriter<User> writer = userDS.write();
        writer.json("output1");

        //4. 關(guān)閉sparkSession
        spark.close();

    }

}

1.3 Parquet文件

列式存儲(chǔ)的數(shù)據(jù)自帶列分割。

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. 創(chuàng)建配置對(duì)象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");

        //2. 獲取sparkSession
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        //3. 編寫(xiě)代碼
        Dataset<Row> json = spark.read().json("input/user.json");
        // 寫(xiě)出默認(rèn)使用snappy壓縮

//        json.write().parquet("output");

        // 讀取parquet 自帶解析  能夠識(shí)別列名
        Dataset<Row> parquet = spark.read().parquet("output");
        parquet.printSchema();

        //4. 關(guān)閉sparkSession
        spark.close();
    }
}

2 與MySQL交互

1）導(dǎo)入依賴

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>

2）從MySQL讀數(shù)據(jù)

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. 創(chuàng)建配置對(duì)象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");

        //2. 獲取sparkSession
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        //3. 編寫(xiě)代碼
        Dataset<Row> json = spark.read().json("input/user.json");

        // 添加參數(shù)

        Properties properties = new Properties();
        properties.setProperty("user","root");
        properties.setProperty("password","000000");

//        json.write()

//                // 寫(xiě)出模式針對(duì)于表格追加覆蓋

//                .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. 關(guān)閉sparkSession
        spark.close();
    }
}

3 與Hive交互

SparkSQL可以采用內(nèi)嵌Hive（spark開(kāi)箱即用的hive），也可以采用外部Hive。企業(yè)開(kāi)發(fā)中，通常采用外部Hive。

3.1 Linux中的交互

1）添加MySQL連接驅(qū)動(dòng)到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）啟動(dòng)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）代碼實(shí)現(xiàn)

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. 創(chuàng)建配置對(duì)象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");

        //2. 獲取sparkSession
        SparkSession spark = SparkSession.builder()
                .enableHiveSupport()// 添加hive支持
                .config(conf).getOrCreate();

        //3. 編寫(xiě)代碼

        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. 關(guān)閉sparkSession
        spark.close();

    }

}

到此這篇關(guān)于深入了解SparkSQL中數(shù)據(jù)的加載與保存的文章就介紹到這了,更多相關(guān)SparkSQL數(shù)據(jù)加載與保存內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: