（一）概述

SparkSQL可以理解為在原生的RDD上做的一層封裝，通過SparkSQL可以在scala和java中寫SQL陳述句，并將結果作為Dataset/DataFrame回傳，簡單來講，SparkSQL可以讓我們像寫SQL一樣去處理記憶體中的資料，

Dataset是一個資料的分布式集合，是Spark1.6之后新增的介面，它提供了RDD的優點和SparkSQL優化執行引擎的優點，一個Dataset相當于RDD+Schema的結合，

Dataset的底層封裝是RDD，當RDD的泛型是Row型別時，該型別就可以稱為DataFrame，DataFrame是一種表格型的資料結構，就和傳統的Mysql結構一樣，通過DataFrame我們可以更加高效地去執行Sql，

（二）SparkSQL實戰

使用SparkSQL首先需要引入相關的依賴：

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

該依賴需要和sparkCore保持一致，

SparkSQL的編碼主要通過四步：

1. 創建SparkSession
2. 獲取資料
3. 執行SQL
4. 關閉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的目錄下創建一個名為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}

運行專案后輸出兩個結果，schema結果如下：

Dataset<Row>輸出結果如下：

通過SparkSQL可以執行和SQL十分相似的查詢操作：

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

在上面的陳述句中，通過一系列的API實作了SQL查詢操作，除此之外，SparkSQL還支持直接寫原始SQL陳述句的操作，

在寫SQL陳述句之前，首先需要讓Spark知道對哪個表進行查詢，因此需要建立一張臨時表，再執行SQL查詢：

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

（三）非JSON格式的Dataset創建

在上一節中創建Dataset時使用了最簡單的json，因為json自己帶有schema結構，因此不需要手動去增加，如果是一個txt檔案，就需要在創建Dataset時手動塞入schema，

下面展示讀取txt檔案的例子，首先創建一個user.txt

a 23
b 24
c 25
d 26

現在我要將上面的這幾行變成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型別轉化為Row型別
        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();
    }
}

（四）通過JDBC創建DataFrame

通過JDBC可直接將對應資料庫中的表放入Spark中進行一些處理，下面通過MySQL進行展示，
使用MySQL需要在依賴中引入MySQL的引擎：

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

接著通過類似JDBC的方式讀取MySQL資料：

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

讀取到的資料是DataFrame，接下來的操作就是對DataFrame的操作了，

（五）總結

SparkSQL是對Spark原生RDD的增強，雖然很多功能通過RDD就可以實作，但是SparkSQL可以更加靈活地實作一些功能，我是魚仔，我們下期再見，