2. 程式人生 > >SparkSQL學習（一）SparkSQL簡單使用

SparkSQL學習（一）SparkSQL簡單使用

阿新 • • 發佈：2019-01-29

1、SparkSQL的進化之路

1.0以前：

      Shark

1.1.x開始：

     SparkSQL(只是測試性的)  SQL

1.3.x:

     SparkSQL(正式版本)+Dataframe

1.5.x:

    SparkSQL 鎢絲計劃

1.6.x：

     SparkSQL+DataFrame+DataSet(測試版本)

1.x:

     SparkSQL+DataFrame+DataSet(正式版本)

     SparkSQL:還有其他的優化

     StructuredStreaming(DataSet)

2、認識SparkSQL

2.1、什麼是SparkSQL

SparkSQL是Spark的一個模組，主要用於進行結構化資料的處理。它提供的最核心的程式設計抽象就是DataFrame。

2.2、SparkSQL的作用

提供一個程式設計抽象（DataFrame）並且作為分散式SQL查詢引擎。

DataFrame：它可以根據很多源進行構建，包括：結構化的資料檔案，Hive中的表，外部的關係型資料庫，以及RDD

2.3、執行原理

將SparkSQL 轉化為 RDD ，然後提交到叢集執行。

2.4、特點

（1）容易整合

（2）統一的資料訪問方式

（3）相容Hive

（4）標準的資料連線

2.5、SparkSession

SparkSession是Spark 2.0引如的新概念。SparkSession為使用者提供了統一的切入點，來讓使用者學習spark的各項功能。 

在spark的早期版本中，SparkContext是spark的主要切入點，由於RDD是主要的API，我們通過sparkcontext來建立和操作RDD。對於每個其他的API，我們需要使用不同的context。例如，對於Streming，我們需要使用StreamingContext；對於sql，使用sqlContext；對於Hive，使用hiveContext。但是隨著DataSet和DataFrame的API逐漸成為標準的API，就需要為他們建立接入點。所以在spark2.0中，引入SparkSession作為DataSet和DataFrame API的切入點，SparkSession封裝了SparkConf、SparkContext和SQLContext。為了向後相容，SQLContext和HiveContext也被儲存下來。 

SparkSession實質上是SQLContext和HiveContext的組合（未來可能還會加上StreamingContext），所以在SQLContext和HiveContext上可用的API在SparkSession上同樣是可以使用的。SparkSession內部封裝了sparkContext，所以計算實際上是由sparkContext完成的。

特點：

　　 ---- 為使用者提供一個統一的切入點使用Spark 各項功能

        ---- 允許使用者通過它呼叫 DataFrame 和 Dataset 相關 API 來編寫程式

        ---- 減少了使用者需要了解的一些概念，可以很容易的與 Spark 進行互動

     ---- 與 Spark 互動之時不需要顯示的建立 SparkConf, SparkContext 以及 SQlContext，這些物件已經封閉在 SparkSession 中

2.6、DataFrames

在Spark中，DataFrame是一種以RDD為基礎的分散式資料集，類似於傳統資料庫中的二維表格。DataFrame與RDD的主要區別在於，前者帶有schema元資訊，即DataFrame所表示的二維表資料集的每一列都帶有名稱和型別。這使得Spark SQL得以洞察更多的結構資訊，從而對藏於DataFrame背後的資料來源以及作用於DataFrame之上的變換進行了針對性的優化，最終達到大幅提升執行時效率的目標，反觀RDD，由於無從得知所存資料元素的具體內部結構，Spark Core只能在stage層面進行簡單、通用的流水線優化。


3、RDD轉換成為DataFrame

使用spark1.x版本的方式

測試資料目錄：/home/hadoop/apps/spark-2.3.0-bin-hadoop2.7/examples/src/main/resources（spark的安裝目錄裡面）

people.txt


3.1、方式一：通過case class 建立 DataFrames（反射）

package Lession2

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

case class People(var name: String,var age:Int)
object TestDataFrame1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDDToDataFrame").setMaster("local")
    val sc = new SparkContext(conf)
    val context = new SQLContext(sc)
    //將本地的資料讀入 RDD ， 並將 RDD 與 case slass關聯
    val peopleRDD: RDD[People] = sc.textFile("D:\\111\\222\\people.txt")
      .map(line => People(line.split(",")(0), line.split(",")(1).trim.toInt))
    
    import context.implicits._
    //將 RDD 轉換成  DataFrames
    val df: DataFrame = peopleRDD.toDF
    //將 DataFrames 建立成一個臨時的檢視
    df.createOrReplaceTempView("people")
    //使用SQL語句進行查詢
    context.sql("select * from people").show()

  }
}

執行結果：


3.2、方式二：通過 structType 建立 DataFrames（程式設計介面）

package Lession2
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object TestDataFrame2 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val fileRDD: RDD[String] = sc.textFile("D:\\111\\222\\people.txt")
    //將 RDD 資料對映成 Row， 需要 import org.apache.spark.sql.Row
    val rowRDD: RDD[Row] = fileRDD.map(line => {
      val fields = line.split(",")
      Row(fields(0), fields(1).trim.toInt)
    })

    //建立 StructType 來定義結構
    val structType: StructType = StructType(
      //欄位名，欄位型別，是否可以為空
      StructField("name", StringType, true) ::
        StructField("age", IntegerType, true) :: Nil
    )

    /**
      * rows: java.util.List[Row]
      * schema: StructType
      */
    val df: DataFrame = sqlContext.createDataFrame(rowRDD,structType)
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()

  }
}

執行結果：


3.3、方式三：通過 json 檔案建立 DataFrames

package Lession2

import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object TestDataFrame3 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("TestDataFrame3 ").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext: SQLContext = new SQLContext(sc)
    val df: DataFrame = sqlContext.read.json("D:\\111\\222\\people.json")
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()

  }
}

執行結果：


4、DataFrame 的 read 和 save 和 savemode

4.1、資料的讀取

package Lession2

import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object TestRead {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("TestRead ").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext: SQLContext = new SQLContext(sc)
    //方式一
    val df1: DataFrame = sqlContext.read.json("D:\\111\\222\\people.json")
    val df2: DataFrame = sqlContext.read.parquet("D:\\111\\222\\users.parquet")
    //方式二
    val df3: DataFrame = sqlContext.read.format("json").load("D:\\111\\222\\people.json")
    val df4: DataFrame = sqlContext.read.format("parquet").load("D:\\111\\222\\users.parquet")
    //方式三，預設是 parquet 模式
    val df5: DataFrame = sqlContext.load("D:\\111\\222\\users.parquet")
    
  }
}

4.2、資料的儲存

package Lession2

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object TestSave {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestSave ").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext: SQLContext = new SQLContext(sc)
    val df1: DataFrame = sqlContext.read.json("D:\\111\\222\\people.json")
    //方式一
//    df1.write.json("D:\\11")
    df1.write.parquet("D:\\11\\2")
    //方式二
    df1.write.format("json").save("D:\\11\\3")
    df1.write.format("parquet").save("D:\\11\\4")
    //方式三
    df1.write.save("D:\\11\\5")
  }
}

4.3、資料的儲存模式

使用mode

df1.write.format("parquet").mode(SaveMode.Ignore).save("E:\\444")

5、資料來源

5.1、資料來源之 json

參考 4.1

5.2　資料來源之parquet

參考4.1

5.3、資料來源之 Mysql

package Lession2

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object TestMysql {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("TestMysql").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    var url = "jdbc:mysql://192.168.123.102:3306/hivedb"
    val table = "DBS"
    val properties = new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","root")

    //需要傳入Mysql的URL、表明、properties（連線資料庫的使用者名稱和密碼）
    val df: DataFrame = sqlContext.read.jdbc(url,table,properties)
    df.createOrReplaceTempView("dbs")
    sqlContext.sql("select * from dbs").show()

  }
}

執行結果：


5.4、資料來源之 Hive

（1）準備工作

在pom.xml檔案中新增依賴

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

開發環境則在 resource 資料夾下新增 hive-site.xml 檔案，叢集環境把 hive 的配置檔案要發到 $SPARK_HOME/conf 目錄下


<configuration>
    <property>
        <name>javax.jdo.option.ConnectionURL</name>
        <value>jdbc:mysql://hadoop02:3306/hivedb?createDatabaseIfNotExist=true</value>
        <description>JDBC connect string for a JDBC metastore</description>
        <!-- 濡傛灉 mysql 鍜hive 鍦ㄥ悓涓€涓湇鍔″櫒鑺傜偣錛岄偅涔堣鏇存敼 hadoop02 涓localhost -->
    </property>
    <property>
        <name>javax.jdo.option.ConnectionDriverName</name>
        <value>com.mysql.jdbc.Driver</value>
        <description>Driver class name for a JDBC metastore</description>
    </property>
    <property>
        <name>javax.jdo.option.ConnectionUserName</name>
        <value>root</value>
        <description>username to use against metastore database</description>
    </property>
    <property>
        <name>javax.jdo.option.ConnectionPassword</name>
        <value>root</value>
        <description>password to use against metastore database</description>
    </property>

    <property>
        <name>hive.metastore.warehouse.dir</name>
        <value>/user/hive/warehouse</value>
        <description>hive default warehouse, if nessecory, change it</description>
    </property>

</configuration>
（2）測試程式碼
object TestHive {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    sqlContext.sql("select * from myhive.student").show()
  }
}

執行結果：


相關推薦

SparkSQL學習（一）SparkSQL簡單使用

1、SparkSQL的進化之路1.0以前：      Shark1.1.x開始：     SparkSQL(只是測試性的)  SQL1.3.x:     SparkSQL(正式版本)+Dataframe1.5.x:    SparkSQL 鎢絲計劃1.6.x：     Spa

[轉]JMeter學習（一）工具簡單介紹

正在 包含 定義 寫入文件 線程組 element 層級 工作臺 roc 一、JMeter 介紹 Apache JMeter是100%純JAVA桌面應用程序，被設計為用於測試客戶端/服務端結構的軟件(例如web應用程序)。它可以用來測試靜態和動態資源的性能，例如：

Unity3D學習（一）：簡單梳理下Unity跨平臺的機制原理

12px get 一個 bsp 嵌入 ram 屬於 開源 runtime 前言 首先需要了解的是，Unity3D的C#基礎腳本模塊是通過Mono來實現的。 什麽是Mono？ 參考下百度百科：Mono是一個由Novell公司（由Xamarin發起）主持的項目，並由Migu

JMeter學習（一）工具簡單介紹

一、JMeter 介紹      Apache JMeter是100%純JAVA桌面應用程式，被設計為用於測試客戶端/服務端結構的軟體(例如web應用程式)。它可以用來測試靜態和動態資源的效能，例如：靜態檔案，Java Servlet,CGI S

SparkSQL學習（三）SparkSQL讀取Mysql的兩種方式

第一種：package Mysql import java.util.Properties import org.apache.spark.sql.DataFrame import org.apach

Tomcat學習筆記（一）一個簡單的Web服務器

sub 調用 [] ont 拒絕 address 剖析 文件 getprop 內容為《深入剖析Tomcat》第一章重點，以及自己的總結，如有描述不清的，可查看原書。 一、HTTP協議： 1、定義：用於服務器與客戶端的通訊的協議，允許web服務器和瀏覽器通過互聯網進行發送和接

.NET中使用Redis之ServiceStack.Redis學習（一）安裝與簡單的運行

arraylist write client cli ring blog 控制臺 創建 spa 1.下載ServiceStack.Redis PM> Install-Package ServiceStack.Redis 2.vs中創建一個控制臺程序 class Pro

JavaScript數據可視化編程學習（一）Flotr2，包含簡單的，柱狀圖，折線圖，餅圖，散點圖

基礎 沒有 cat 勝利 而是 5.4 最好的 表數據 聯系 一、基礎柱狀圖 二、基礎的折線圖 三、基礎的餅圖 四、基礎的散點圖 一、基礎柱狀圖 如果你還沒有想好你的數據用什麽類型的圖表來展示你的數據，你應該首先考慮是否可以做成柱狀圖。柱狀圖可以表示數據的

一起學習Rviz（一）——顯示簡單立體模型

edit ide tutorials setup tutorial 添加 tar dev als 先甩個鍋咯，手敲下來的難免有些小錯誤，錯別字啥的，理解哈~~~ Ubuntu 14.04，ROS版本 indigo 1、Rviz顯示簡單立體模型 首先，創建工作空間。 創建文件

CSS簡單學習（一）-CSS介紹和CSS引入方式

CSS作用和發展 CSS（Cascading Style Sheet），稱之為層疊樣式表 一種專門描述結構文件的變現方式文件，主要用於網頁風格設計，包括字型大小，顏色，以及元素的精確定位等。 在傳統的網頁設計裡，使用CSS能讓單調的HTML網頁更富表現力 作用：

Oracle資料庫學習（一）安裝和簡單使用

新公司的新專案，需要用到Oracle資料庫，所以現在便來解除此資料庫，不得不說，這個資料庫還這是麻煩。 安裝倒是簡單，就是中間會遇到各種問題。 安裝步驟參考：https://blog.csdn.net/fancheng614/article/details/78172894 安裝完成後，中間不要有太大的

ubuntu18.04安裝gitlab-ee11.4.6及簡單學習（一）

參考：https://www.gitlab.com.cn/installation/#ubuntu 今天是2018年11月22日 Gitlab-ee最新版本是11.4.6 採用ubuntu18.04安裝gitlab-ee 官方說了，如果不確定裝gitlab-ce還是gitlab-

MATLAB學習之路（一） 實現簡單的基於歐式距離的新型聚類演算法（Clustering by fast search and find of density peaksd）

大學本科三年，眨眼而已，對於一個考研黨來說，本科時間已所剩不多，大三上學期初次接觸到MATLAB。的確如大牛們所說，強大的計算能力，充足的數學工具，帶來使用者極大的便利。 在大資料的學習過程中，MAT

tensorflow學習（一）——有關tensorflow不同層的使用（tf.nn 和tf.layers以及tf.contrib.layers）的簡單區別

小trick: 對於使用tf.layers建立的神經網路，如果想要對loss函式進行正則話，可以採用如下方式[1]： 但是該方法不適用於程式設計者自己定義不同層的正則化。 l2 = tf.add_n([tf.nn.l2_loss(var) for var in tf.t

SpringMVC學習（一）———— springmvc框架原理分析和簡單入門程式

一、什麼是springmvc？ 　　　　　　我們知道三層架構的思想，並且如果你知道ssh的話，就會更加透徹的理解這個思想，struts2在web層，spring在中間控制，hibernate在dao層與資料庫打交道，而前面剛寫的mybatis跟hibernate一樣，與資料庫打交道在dao層的另一個框架，而

apache HttpClient學習（一）——簡單的post請求

import org.apache.http.*; import org.apache.http.client.HttpClient; import org.apache.http.client.entity.UrlEncodedFormEntity; import org.apache.http.clien

MyBatis學習筆記（一）---一個簡單MyBatis示例

public class Test { private static SqlSessionFactory sqlSessionFactory= MyBatisUtil.getSqlSessionFactory(); public static void main(String args[]

ECharts的學習（一）：製作一個簡單的柱狀圖

<!DOCTYPE html> <html> <head> <meta charset="utf-8"><meta na

lucene簡單的學習（一）

今天小白剛學lucene，記錄下自己的學習。先上程式碼，然後再分析 package com.wm.util; import java.io.*; import java.util.ArrayList; import java.util.Date; import java.util.List

Hessian學習之（一）：簡單測試 + C#呼叫Java服務

開始調研Hessian和Mina，目的是希望能夠建立一個簡單的“高效能分散式服務呼叫框架”！類似於Dubbo或者淘寶的HSF那種，但是是要跨平臺的，而不僅僅侷限在Java領域。 hessian是一種遠端呼叫的機制（RPC） ，類似於web service，不過它是使用自己