SparkSQL介紹

Hive是Shark的前身，Shark是SparkSQL的前身,SparkSQL產生的根本原因是其完全脫離了Hive的限制。

SparkSQL支持查詢原生的RDD。 RDD是Spark平臺的核心概念，是Spark能夠高效的處理大數據的各種場景的基礎。

能夠在scala中寫SQL語句。支持簡單的SQL語法檢查，能夠在Scala中寫Hive語句訪問Hive數據，并將結果取回作為RDD使用。

Spark on Hive和Hive on Spark

Spark on Hive： Hive只作為儲存角色，Spark負責sql解析優化，執行。

Hive on Spark：Hive即作為存儲又負責sql的解析優化，Spark負責執行。

Dataset與DataFrame

Dataset也是一個分布式數據容器。與RDD類似，然而Dataset更像傳統數據庫的二維表格，除了數據以外，還掌握數據的結構信息，即schema。同時，與Hive類似，Dataset也支持嵌套數據類型（struct、array和map）。從API易用性的角度上 看， Dataset API提供的是一套高層的關系操作，比函數式的RDD API要更加友好，門檻更低。

Dataset的底層封裝的是RDD，當RDD的泛型是Row類型的時候，我們也可以稱它為DataFrame。即Dataset<Row> = DataFrame

SparkSQL底層架構

首先拿到sql后解析一批未被解決的邏輯計劃，再經過分析得到分析后的邏輯計劃，再經過一批優化規則轉換成一批最佳優化的邏輯計劃，再經過SparkPlanner的策略轉化成一批物理計劃，隨后經過消費模型轉換成一個個的Spark任務執行。

謂詞下推（predicate Pushdown）

創建Dataset的幾種方式

讀取json格式的文件創建Dataset

注意：

• json文件中的json數據不能嵌套json格式數據。
• Dataset是一個一個Row類型的RDD，ds.rdd()/ds.javaRdd()。
• 可以兩種方式讀取json格式的文件。
• df.show()默認顯示前20行數據。
• Dataset原生API可以操作Dataset（不方便）。
• 注冊成臨時表時，表中的列默認按ascii順序顯示列。

java: SparkSession sparkSession = SparkSession.builder().appName("jsonfile").master("local").getOrCreate(); /*** Dataset的底層是一個一個的RDD RDD的泛型是Row類型。* 以下兩種方式都可以讀取json格式的文件*/ Dataset<Row> ds = sparkSession.read().format("json").load("data/json"); // Dataset<Row> ds = sparkSession.read().json("data/json"); // ds.show();/*** Dataset轉換成RDD*/ JavaRDD<Row> javaRDD = ds.javaRDD(); /*** 顯示 Dataset中的內容，默認顯示前20行。如果現實多行要指定多少行show(行數)* 注意：當有多個列時，顯示的列先后順序是按列的ascii碼先后顯示。*/ // ds.show(); /*** 樹形的形式顯示schema信息*/ds.printSchema();/*** Dataset自帶的API 操作Dataset*///select name from table// ds.select("name").show();//select name age+10 as addage from tableds.select(ds.col("name"),ds.col("age").plus(10).alias("addage")).show();//select name ,age from table where age>19ds.select(ds.col("name"),ds.col("age")).where(ds.col("age").gt(19)).show();//select count(*) from table group by ageds.groupBy(ds.col("age")).count().show();/*** 將Dataset注冊成臨時的一張表，這張表臨時注冊到內存中，是邏輯上的表，不會霧化到磁盤*/ds.createOrReplaceTempView("jtable");Dataset<Row> result= sparkSession.sql("select age,count(*) as gg from jtable group by age");result.show(); sparkSession.stop();

通過json格式的RDD創建Dataset

java: SparkSession sparkSession = SparkSession.builder().appName("jsonrdd").master("local").getOrCreate(); SparkContext sc = sparkSession.sparkContext(); JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList("{"name":"zhangsan","age":"18"}","{"name":"lisi","age":"19"}","{"name":"wangwu","age":"20"}" )); JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList( "{"name":"zhangsan","score":"100"}", "{"name":"lisi","score":"200"}", "{"name":"wangwu","score":"300"}" ));Dataset<Row> nameds = sparkSession.read().json(nameRDD); Dataset<Row> scoreds = sparkSession.read().json(scoreRDD); nameds.createOrReplaceTempView("nameTable"); scoreds.createOrReplaceTempView("scoreTable"); Dataset<Row> result =sparkSession.sql("select nameTable.name,nameTable.age,scoreTable.score "+ "from nameTable join scoreTable "+ "on nameTable.name = scoreTable.name"); result.show(); sparkSession.stop();非json格式的RDD創建Dataset 通過反射的方式將非json格式的RDD轉換成Dataset 自定義類要可序列化 自定義類的訪問級別是Public RDD轉成Dataset后會根據映射將字段按Assci碼排序 將Dataset轉換成RDD時獲取字段兩種方式,一種是ds.getInt(0)下標獲取（不推薦使用）， 另一種是ds.getAs(“列名”)獲取（推薦使用） /** * 注意： * 1.自定義類必須是可序列化的 * 2.自定義類訪問級別必須是Public * 3.RDD轉成Dataset會把自定義類中字段的名稱按assci碼排序 */ SparkSession sparkSession = SparkSession.builder().appName("reflect").master("local").getOrCreate();SparkContext sc = sparkSession.sparkContext();JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> lineRDD = jsc.textFile("data/person.txt");JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() {private static final long serialVersionUID = 1L;@Overridepublic Person call(String line) throws Exception {Person p = new Person();p.setId(line.split(",")[0]);p.setName(line.split(",")[1]);p.setAge(Integer.valueOf(line.split(",")[2]));return p;} }); /** * 傳入進去Person.class的時候，sqlContext是通過反射的方式創建DataFrame * 在底層通過反射的方式獲得Person的所有field，結合RDD本身，就生成了DataFrame */ Dataset<Row> dataFrame = sparkSession.createDataFrame(personRDD, Person.class);dataFrame.show();dataFrame.printSchema();dataFrame.registerTempTable("person"); Dataset sql = sparkSession.sql("select name,id,age from person where id = 2");sql.show();/** * 將Dataset轉成JavaRDD * 注意： * 1.可以使用row.getInt(0),row.getString(1)...通過下標獲取返回Row類型的數據， 但是要注意列順序問題---不常用 * 2.可以使用row.getAs("列名")來獲取對應的列值。 * */ JavaRDD<Row> javaRDD = df.javaRDD(); JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Person call(Row row) throws Exception {Person p = new Person();//p.setId(row.getString(1));//p.setName(row.getString(2));//p.setAge(row.getInt(0));p.setId((String)row.getAs("id"));p.setName((String)row.getAs("name"));p.setAge((Integer)row.getAs("age"));return p;} }); map.foreach(x-> System.out.println(x));sc.stop();

動態創建Schema將非json格式的RDD轉換成Dataset

java： SparkSession sparkSession = SparkSession.builder().appName("schema").master("local").getOrCreate();SparkContext sc = sparkSession.sparkContext(); JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> lineRDD = jsc.textFile("data/person.txt"); JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt"); /*** 轉換成Row類型的RDD*/ JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(String.valueOf(s.split(",")[0]),String.valueOf(s.split(",")[1]),Integer.valueOf(s.split(",")[2]));} }); /*** 動態構建DataFrame中的元數據，一般來說這里的字段可以來源自字符串，也可以來源于外部數據庫*/ List<StructField> asList =Arrays.asList(DataTypes.createStructField("id", DataTypes.StringType, true),DataTypes.createStructField("name", DataTypes.StringType, true),DataTypes.createStructField("age", DataTypes.IntegerType, true) );StructType schema = DataTypes.createStructType(asList); Dataset<Row> df = sparkSession.createDataFrame(rowRDD, schema); df.show(); sc.stop();

讀取parquet文件創建Dataset

注意：

• 可以將Dataset存儲成parquet文件。保存成parquet文件的方式有兩種

df.write().mode(SaveMode.Overwrite)format("parquet")

.save("./sparksql/parquet");

df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");

• SaveMode指定文件保存時的模式。

Overwrite：覆蓋

Append：追加

ErrorIfExists：如果存在就報錯

Ignore：如果存在就忽略

java:

SparkSession sparkSession = SparkSession.builder().appName("parquet").master("local").getOrCreate();Dataset<Row> df = sparkSession.read().json("data/json"); /*** 將Dataset保存成parquet文件，SaveMode指定存儲文件時的保存模式* 保存成parquet文件有以下兩種方式：*/ df.write().mode(SaveMode.Overwrite).format("parquet").save("./data/parquet"); df.write().mode(SaveMode.Overwrite).parquet("./data/parquet"); df.show(); /*** 加載parquet文件成DataFrame * 加載parquet文件有以下兩種方式： */Dataset load = sparksession.read().format("parquet").load("./data /parquet");load = sparksession.read().parquet("./data /parquet"); load.show();sc.stop();

讀取JDBC中的數據創建Dataset(MySql為例)

兩種方式創建Dataset

java:

SparkSession sparkSession = SparkSession.builder().appName("mysql").master("local").getOrCreate();/*** 第一種方式讀取MySql數據庫表，加載為DataFrame*/ Map<String, String> options = new HashMap<String,String>(); options.put("url", "jdbc:mysql://192.168.179.4:3306/spark"); options.put("driver", "com.mysql.jdbc.Driver"); options.put("user", "root"); options.put("password", "123456"); options.put("dbtable", "person"); Dataset<Row> person = sparkSession.read().format("jdbc").options(options).load(); person.show(); person.createOrReplaceTempView("person"); /*** 第二種方式讀取MySql數據表加載為Dataset*/ DataFrameReader reader = sparkSession.read().format("jdbc"); reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark"); reader.option("driver", "com.mysql.jdbc.Driver"); reader.option("user", "root"); reader.option("password", "123456"); reader.option("dbtable", "score"); Dataset<Row> score = reader.load(); score.show(); score.createOrReplaceTempView("score"); Dataset result = sparksession.sql("select person.id,person.name,score.score from person,score where person.name = score.name"); result.show(); /*** 將Dataset結果保存到Mysql中*/ Properties properties = new Properties(); properties.setProperty("user", "root"); properties.setProperty("password", "root"); result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties);sc.stop();

讀取Hive中的數據加載成Dataset

java:

SparkSession sparkSession = SparkSession.builder().master("local").appName("hvie")//開啟hive的支持，接下來就可以操作hive表了// 前提需要是需要開啟hive metastore 服務.enableHiveSupport().getOrCreate();sparkSession.sql("USE spark");sparkSession.sql("DROP TABLE IF EXISTS student_infos"); //在hive中創建student_infos表sparkSession.sql("CREATE TABLE IF NOT EXISTS student_infos (name STRING,age INT) row format delimited fields terminated by 't' ");sparkSession.sql("load data local inpath '/root/student_infos' into table student_infos");//注意：此種方式，程序需要能讀取到數據（如/root/student_infos），同時也要能讀取到 metastore服務的配置信息。sparkSession.sql("DROP TABLE IF EXISTS student_scores"); sparkSession.sql("CREATE TABLE IF NOT EXISTS student_scores (name STRING, score INT) row format delimited fields terminated by 't'");sparkSession.sql("LOAD DATA "+ "LOCAL INPATH '/root/student_scores'"+ "INTO TABLE student_scores");// Dataset<Row> df = hiveContext.table("student_infos");//讀取Hive表加載Dataset方式/*** 查詢表生成Dataset*/Dataset<Row> goodStudentsDF = sparkSession.sql("SELECT si.name, si.age, ss.score "+ "FROM student_infos si "+ "JOIN student_scores ss "+ "ON si.name=ss.name "+ "WHERE ss.score>=80");goodStudentsDF.registerTempTable("goodstudent");Dataset<Row> result = sparkSession.sql("select * from goodstudent");result.show();/*** 將結果保存到hive表 good_student_infos*/sparkSession.sql("DROP TABLE IF EXISTS good_student_infos"); goodStudentsDF.write().mode(SaveMode.Overwrite).saveAsTable("good_student_infos");sparkSession.stop();

Spark On Hive的配置

在Spark客戶端配置spark On hive

在Spark客戶端安裝包下spark-2.2.1/conf中創建文件hive-site.xml：

配置hive的metastore路徑

<configuration>

<property>

<name>hive.metastore.uris</name>

<value>thrift://node1:9083</value>

</property>

</configuration>

啟動Hive的metastore服務

hive --service metastore &

啟動zookeeper集群，啟動HDFS集群。

啟動SparkShell 讀取Hive中的表總數，對比hive中查詢同一表查詢總數測試時間。

./spark-shell

--master spark://node1:7077

spark.sql("select * from day_table").show;

• 注意：

如果使用Spark on Hive 查詢數據時，出現錯誤：

找不到HDFS集群路徑，要在客戶端機器conf/spark-env.sh中設置HDFS的路徑：

序列化問題。

儲存DataSet

將DataSet存儲為parquet文件。

將DataSet存儲到JDBC數據庫。

將DataSet存儲到Hive表。

自定義函數UDF和UDAF

UDF:用戶自定義函數。

java:

SparkSession sparkSession = SparkSession.builder().appName("udf").master("local").getOrCreate();JavaSparkContext sc = new JavaSparkContext(sparkSession.sparkContext()); JavaRDD<String> parallelize = sc.parallelize(Arrays.asList("zhangsan","lisi","wangwu"));JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(s);} });/*** 動態創建Schema方式加載DF*/ List<StructField> fields = new ArrayList<StructField>(); fields.add(DataTypes.createStructField("name", DataTypes.StringType,true)); StructType schema = DataTypes.createStructType(fields);Dataset<Row> df = sparkSession.createDataFrame(rowRDD,schema);df.registerTempTable("user");/*** 根據UDF函數參數的個數來決定是實現哪一個UDF UDF1，UDF2。。。。UDF1xxx*/sparkSession.udf().register("StrLen",new UDF2<String, Integer, Integer>() {/****/private static final long serialVersionUID = 1L;@Overridepublic Integer call(String t1, Integer t2) throws Exception {return t1.length() + t2;}} ,DataTypes.IntegerType );sparkSession.sql("select name ,StrLen(name,100) as length from user").show(); // sparkSession.stop();

UDAF:用戶自定義聚合函數。

• 實現UDAF函數如果要自定義類要實現UserDefinedAggregateFunction類

java:

SparkSession sparkSession = SparkSession.builder().appName("udaf").master("local").getOrCreate();JavaSparkContext sc = new JavaSparkContext(sparkSession.sparkContext()); JavaRDD<String> parallelize = sc.parallelize(Arrays.asList("zhansan","lisi","wangwu","zhangsan","zhangsan","lisi")); JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(s);} });List<StructField> fields = new ArrayList<StructField>(); fields.add(DataTypes.createStructField("name", DataTypes.StringType, true)); StructType schema = DataTypes.createStructType(fields); Dataset df = sparkSession.createDataFrame(rowRDD, schema); df.registerTempTable("user"); /*** 注冊一個UDAF函數,實現統計相同值得個數* 注意：這里可以自定義一個類繼承UserDefinedAggregateFunction類也是可以的*/ sparkSession.udf().register("StringCount", new UserDefinedAggregateFunction() {/*** */private static final long serialVersionUID = 1L;/*** 更新 可以認為一個一個地將組內的字段值傳遞進來 實現拼接的邏輯* buffer.getInt(0)獲取的是上一次聚合后的值* 相當于map端的combiner，combiner就是對每一個map task的處理結果進行一次小聚合 * 大聚和發生在reduce端.* 這里即是:在進行聚合的時候，每當有新的值進來，對分組后的聚合如何進行計算*/@Overridepublic void update(MutableAggregationBuffer buffer, Row arg1) {buffer.update(0, buffer.getInt(0)+1);}/*** 合并 update操作，可能是針對一個分組內的部分數據，在某個節點上發生的 但是可能一個分組內的數據，會分布在多個節點上處理* 此時就要用merge操作，將各個節點上分布式拼接好的串，合并起來* buffer1.getInt(0) : 大聚和的時候 上一次聚合后的值 * buffer2.getInt(0) : 這次計算傳入進來的update的結果* 這里即是：最后在分布式節點完成后需要進行全局級別的Merge操作*/@Overridepublic void merge(MutableAggregationBuffer buffer1, Row buffer2) {buffer1.update(0, buffer1.getInt(0) + buffer2.getInt(0));}/*** 指定輸入字段的字段及類型*/@Overridepublic StructType inputSchema() {return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("name", DataTypes.StringType, true)));}/*** 初始化一個內部的自己定義的值,在Aggregate之前每組數據的初始化結果*/@Overridepublic void initialize(MutableAggregationBuffer buffer) {buffer.update(0, 0);}/*** 最后返回一個和DataType的類型要一致的類型，返回UDAF最后的計算結果*/@Overridepublic Object evaluate(Row row) {return row.getInt(0);}@Overridepublic boolean deterministic() {//設置為truereturn true;}/*** 指定UDAF函數計算后返回的結果類型*/@Overridepublic DataType dataType() {return DataTypes.IntegerType;}/*** 在進行聚合操作的時候所要處理的數據的結果的類型*/@Overridepublic StructType bufferSchema() {return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("bf", DataTypes.IntegerType, true)));}});sparkSession.sql("select name ,StringCount(name) from user group by name").show();sparkSession.stop();

開窗函數

注意：

row_number() 開窗函數是按照某個字段分組，然后取另一字段的前幾個的值，相當于 分組取topN

開窗函數格式：

row_number() over (partitin by XXX order by XXX)

java:

SparkSession sparkSession = SparkSession.builder().appName("window").master("local")//開啟hive的支持，接下來就可以操作hive表了// 前提需要是需要開啟hive metastore 服務.enableHiveSupport().getOrCreate();sparkSession.sql("use spark");sparkSession.sql("drop table if exists sales");sparkSession.sql("create table if not exists sales (riqi string,leibie string,jine Int) "+ "row format delimited fields terminated by 't'");sparkSession.sql("load data local inpath '/root/test/sales' into table sales");/*** 開窗函數格式：* 【 rou_number() over (partitin by XXX order by XXX) 】*/Dataset<Row> result = sparkSession.sql("select riqi,leibie,jine "+ "from ("+ "select riqi,leibie,jine,"+ "row_number() over (partition by leibie order by jine desc) rank "+ "from sales) t "+ "where t.rank<=3");result.show();sparkSession.stop();

總結

以上是生活随笔為你收集整理的spark-sql建表语句限制_SparkSQL的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。