快捷導(dǎo)航

SparkSQL快速入門(mén)教程

更新時(shí)間：2022年02月08日 14:51:28 作者：Java魚(yú)仔

Spark?SQL是Spark用于結(jié)構(gòu)化數(shù)據(jù)(structured?data)處理的Spark模塊，今天通過(guò)本文給大家介紹SparkSQL快速入門(mén)教程，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧

（一）概述

SparkSQL可以理解為在原生的RDD上做的一層封裝，通過(guò)SparkSQL可以在scala和java中寫(xiě)SQL語(yǔ)句，并將結(jié)果作為Dataset/DataFrame返回。簡(jiǎn)單來(lái)講，SparkSQL可以讓我們像寫(xiě)SQL一樣去處理內(nèi)存中的數(shù)據(jù)。

Dataset是一個(gè)數(shù)據(jù)的分布式集合，是Spark1.6之后新增的接口，它提供了RDD的優(yōu)點(diǎn)和SparkSQL優(yōu)化執(zhí)行引擎的優(yōu)點(diǎn)，一個(gè)Dataset相當(dāng)于RDD+Schema的結(jié)合。

Dataset的底層封裝是RDD，當(dāng)RDD的泛型是Row類(lèi)型時(shí)，該類(lèi)型就可以稱(chēng)為DataFrame。DataFrame是一種表格型的數(shù)據(jù)結(jié)構(gòu)，就和傳統(tǒng)的Mysql結(jié)構(gòu)一樣，通過(guò)DataFrame我們可以更加高效地去執(zhí)行Sql。

特點(diǎn)

易整合，在程序中既可以使用SQL，還可以使用API！
統(tǒng)一的數(shù)據(jù)訪(fǎng)問(wèn)，不同數(shù)據(jù)源中的數(shù)據(jù)，都可以使用SQL或DataFrameAPI進(jìn)行操作，還可以進(jìn)行不同數(shù)據(jù)源的Join！
對(duì)Hive的無(wú)縫支持
支持標(biāo)準(zhǔn)的JDBC和ODBC

（二）SparkSQL實(shí)戰(zhàn)

使用SparkSQL首先需要引入相關(guān)的依賴(lài)：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.0.0</version>
</dependency>

該依賴(lài)需要和sparkCore保持一致。

SparkSQL的編碼主要通過(guò)四步：

創(chuàng)建SparkSession
獲取數(shù)據(jù)
執(zhí)行SQL
關(guān)閉SparkSession

public class SqlTest {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder()
                .appName("sql")
                .master("local")
                .getOrCreate();
        Dataset<Row> json = sparkSession.read().json("data/json");
        json.printSchema();
        json.show();
        sparkSession.stop();
    }
}

在data的目錄下創(chuàng)建一個(gè)名為json的文件

{"name":"a","age":23}
{"name":"b","age":24}
{"name":"c","age":25}
{"name":"d","age":26}
{"name":"e","age":27}
{"name":"f","age":28}

運(yùn)行項(xiàng)目后輸出兩個(gè)結(jié)果，schema結(jié)果如下：

Dataset<Row>輸出結(jié)果如下：

通過(guò)SparkSQL可以執(zhí)行和SQL十分相似的查詢(xún)操作：

public class SqlTest {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder()
                .appName("sql")
                .master("local")
                .getOrCreate();
        Dataset<Row> json = sparkSession.read().json("data/json");
        json.select("age","name").where("age > 26").show();
        sparkSession.stop();
    }
}

在上面的語(yǔ)句中，通過(guò)一系列的API實(shí)現(xiàn)了SQL查詢(xún)操作，除此之外，SparkSQL還支持直接寫(xiě)原始SQL語(yǔ)句的操作。

在寫(xiě)SQL語(yǔ)句之前，首先需要讓Spark知道對(duì)哪個(gè)表進(jìn)行查詢(xún)，因此需要建立一張臨時(shí)表，再執(zhí)行SQL查詢(xún)：

json.createOrReplaceTempView("json");
sparkSession.sql("select * from json where age > 26").show();

（三）非JSON格式的Dataset創(chuàng)建

在上一節(jié)中創(chuàng)建Dataset時(shí)使用了最簡(jiǎn)單的json，因?yàn)閖son自己帶有schema結(jié)構(gòu)，因此不需要手動(dòng)去增加，如果是一個(gè)txt文件，就需要在創(chuàng)建Dataset時(shí)手動(dòng)塞入schema。

下面展示讀取txt文件的例子，首先創(chuàng)建一個(gè)user.txt

a 23
b 24
c 25
d 26

現(xiàn)在我要將上面的這幾行變成DataFrame，第一列表示姓名，第二列表示年齡，于是就可以像下面這樣操作：

public class SqlTest2 {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder()
                .appName("sql")
                .master("local")
                .getOrCreate();
        SparkContext sparkContext = sparkSession.sparkContext();
        JavaSparkContext sc = new JavaSparkContext(sparkContext);
        JavaRDD<String> lines = sc.textFile("data/user.txt");
        //將String類(lèi)型轉(zhuǎn)化為Row類(lèi)型
        JavaRDD<Row> rowJavaRDD = lines.map(new Function<String, Row>() {
            @Override
            public Row call(String v1) throws Exception {
                String[] split = v1.split(" ");
                return RowFactory.create(
                        split[0],
                        Integer.valueOf(split[1])
                );
            }
        });
        //定義schema
        List<StructField> structFields = Arrays.asList(
                DataTypes.createStructField("name", DataTypes.StringType, true),
                DataTypes.createStructField("age", DataTypes.IntegerType, true)
        );
        StructType structType = DataTypes.createStructType(structFields);
        //生成dataFrame
        Dataset<Row> dataFrame = sparkSession.createDataFrame(rowJavaRDD, structType);
        dataFrame.show();
    }
}

（四）通過(guò)JDBC創(chuàng)建DataFrame

通過(guò)JDBC可直接將對(duì)應(yīng)數(shù)據(jù)庫(kù)中的表放入Spark中進(jìn)行一些處理，下面通過(guò)MySQL進(jìn)行展示。
使用MySQL需要在依賴(lài)中引入MySQL的引擎：

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

接著通過(guò)類(lèi)似JDBC的方式讀取MySQL數(shù)據(jù)：

public class SqlTest3 {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder()
                .appName("sql")
                .master("local")
                .getOrCreate();
        Map<String,String> options = new HashMap<>();
        options.put("url","jdbc:mysql://127.0.0.1:3306/books");
        options.put("driver","com.mysql.jdbc.Driver");
        options.put("user","root");
        options.put("password","123456");
        options.put("dbtable","book");
        Dataset<Row> jdbc = sparkSession.read().format("jdbc").options(options).load();
        jdbc.show();
        sparkSession.close();
    }
}

讀取到的數(shù)據(jù)是DataFrame，接下來(lái)的操作就是對(duì)DataFrame的操作了。