SparkSQL介紹

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

SparkSQL支持查詢?cè)腞DD。 RDD是Spark平臺(tái)的核心概念，是Spark能夠高效的處理大數(shù)據(jù)的各種場(chǎng)景的基礎(chǔ)。

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

Spark on Hive和Hive on Spark

Spark on Hive： Hive只作為儲(chǔ)存角色，Spark負(fù)責(zé)sql解析優(yōu)化，執(zhí)行。

Hive on Spark：Hive即作為存儲(chǔ)又負(fù)責(zé)sql的解析優(yōu)化，Spark負(fù)責(zé)執(zhí)行。

Dataset與DataFrame

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

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

SparkSQL底層架構(gòu)

首先拿到sql后解析一批未被解決的邏輯計(jì)劃，再經(jīng)過分析得到分析后的邏輯計(jì)劃，再經(jīng)過一批優(yōu)化規(guī)則轉(zhuǎn)換成一批最佳優(yōu)化的邏輯計(jì)劃，再經(jīng)過SparkPlanner的策略轉(zhuǎn)化成一批物理計(jì)劃，隨后經(jīng)過消費(fèi)模型轉(zhuǎn)換成一個(gè)個(gè)的Spark任務(wù)執(zhí)行。

謂詞下推（predicate Pushdown）

創(chuàng)建Dataset的幾種方式

讀取json格式的文件創(chuàng)建Dataset

注意：

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

java: SparkSession sparkSession = SparkSession.builder().appName("jsonfile").master("local").getOrCreate(); /*** Dataset的底層是一個(gè)一個(gè)的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轉(zhuǎn)換成RDD*/ JavaRDD<Row> javaRDD = ds.javaRDD(); /*** 顯示 Dataset中的內(nèi)容，默認(rèn)顯示前20行。如果現(xiàn)實(shí)多行要指定多少行show(行數(shù))* 注意：當(dāng)有多個(gè)列時(shí)，顯示的列先后順序是按列的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注冊(cè)成臨時(shí)的一張表，這張表臨時(shí)注冊(cè)到內(nèi)存中，是邏輯上的表，不會(huì)霧化到磁盤*/ds.createOrReplaceTempView("jtable");Dataset<Row> result= sparkSession.sql("select age,count(*) as gg from jtable group by age");result.show(); sparkSession.stop();

通過json格式的RDD創(chuàng)建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創(chuàng)建Dataset 通過反射的方式將非json格式的RDD轉(zhuǎn)換成Dataset 自定義類要可序列化 自定義類的訪問級(jí)別是Public RDD轉(zhuǎn)成Dataset后會(huì)根據(jù)映射將字段按Assci碼排序 將Dataset轉(zhuǎn)換成RDD時(shí)獲取字段兩種方式,一種是ds.getInt(0)下標(biāo)獲取（不推薦使用）， 另一種是ds.getAs(“列名”)獲取（推薦使用） /** * 注意： * 1.自定義類必須是可序列化的 * 2.自定義類訪問級(jí)別必須是Public * 3.RDD轉(zhuǎn)成Dataset會(huì)把自定義類中字段的名稱按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;} }); /** * 傳入進(jìn)去Person.class的時(shí)候，sqlContext是通過反射的方式創(chuàng)建DataFrame * 在底層通過反射的方式獲得Person的所有field，結(jié)合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轉(zhuǎn)成JavaRDD * 注意： * 1.可以使用row.getInt(0),row.getString(1)...通過下標(biāo)獲取返回Row類型的數(shù)據(jù)， 但是要注意列順序問題---不常用 * 2.可以使用row.getAs("列名")來獲取對(duì)應(yīng)的列值。 * */ 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();

動(dòng)態(tài)創(chuàng)建Schema將非json格式的RDD轉(zhuǎn)換成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"); /*** 轉(zhuǎn)換成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]));} }); /*** 動(dòng)態(tài)構(gòu)建DataFrame中的元數(shù)據(jù)，一般來說這里的字段可以來源自字符串，也可以來源于外部數(shù)據(jù)庫*/ 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文件創(chuàng)建Dataset

注意：

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

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

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

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

• SaveMode指定文件保存時(shí)的模式。

Overwrite：覆蓋

Append：追加

ErrorIfExists：如果存在就報(bào)錯(cuò)

Ignore：如果存在就忽略

java:

SparkSession sparkSession = SparkSession.builder().appName("parquet").master("local").getOrCreate();Dataset<Row> df = sparkSession.read().json("data/json"); /*** 將Dataset保存成parquet文件，SaveMode指定存儲(chǔ)文件時(shí)的保存模式* 保存成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中的數(shù)據(jù)創(chuàng)建Dataset(MySql為例)

兩種方式創(chuàng)建Dataset

java:

SparkSession sparkSession = SparkSession.builder().appName("mysql").master("local").getOrCreate();/*** 第一種方式讀取MySql數(shù)據(jù)庫表，加載為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數(shù)據(jù)表加載為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結(jié)果保存到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中的數(shù)據(jù)加載成Dataset

java:

SparkSession sparkSession = SparkSession.builder().master("local").appName("hvie")//開啟hive的支持，接下來就可以操作hive表了// 前提需要是需要開啟hive metastore 服務(wù).enableHiveSupport().getOrCreate();sparkSession.sql("USE spark");sparkSession.sql("DROP TABLE IF EXISTS student_infos"); //在hive中創(chuàng)建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");//注意：此種方式，程序需要能讀取到數(shù)據(jù)（如/root/student_infos），同時(shí)也要能讀取到 metastore服務(wù)的配置信息。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();/*** 將結(jié)果保存到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中創(chuàng)建文件hive-site.xml：

配置hive的metastore路徑

<configuration>

<property>

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

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

</property>

</configuration>

啟動(dòng)Hive的metastore服務(wù)

hive --service metastore &

啟動(dòng)zookeeper集群，啟動(dòng)HDFS集群。

啟動(dòng)SparkShell 讀取Hive中的表總數(shù)，對(duì)比hive中查詢同一表查詢總數(shù)測(cè)試時(shí)間。

./spark-shell

--master spark://node1:7077

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

• 注意：

如果使用Spark on Hive 查詢數(shù)據(jù)時(shí)，出現(xiàn)錯(cuò)誤：

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

序列化問題。

儲(chǔ)存DataSet

將DataSet存儲(chǔ)為parquet文件。

將DataSet存儲(chǔ)到JDBC數(shù)據(jù)庫。

將DataSet存儲(chǔ)到Hive表。

自定義函數(shù)UDF和UDAF

UDF:用戶自定義函數(shù)。

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);} });/*** 動(dòng)態(tài)創(chuàng)建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");/*** 根據(jù)UDF函數(shù)參數(shù)的個(gè)數(shù)來決定是實(shí)現(xiàn)哪一個(gè)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:用戶自定義聚合函數(shù)。

• 實(shí)現(xiàn)UDAF函數(shù)如果要自定義類要實(shí)現(xiàn)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"); /*** 注冊(cè)一個(gè)UDAF函數(shù),實(shí)現(xiàn)統(tǒng)計(jì)相同值得個(gè)數(shù)* 注意：這里可以自定義一個(gè)類繼承UserDefinedAggregateFunction類也是可以的*/ sparkSession.udf().register("StringCount", new UserDefinedAggregateFunction() {/*** */private static final long serialVersionUID = 1L;/*** 更新 可以認(rèn)為一個(gè)一個(gè)地將組內(nèi)的字段值傳遞進(jìn)來 實(shí)現(xiàn)拼接的邏輯* buffer.getInt(0)獲取的是上一次聚合后的值* 相當(dāng)于map端的combiner，combiner就是對(duì)每一個(gè)map task的處理結(jié)果進(jìn)行一次小聚合 * 大聚和發(fā)生在reduce端.* 這里即是:在進(jìn)行聚合的時(shí)候，每當(dāng)有新的值進(jìn)來，對(duì)分組后的聚合如何進(jìn)行計(jì)算*/@Overridepublic void update(MutableAggregationBuffer buffer, Row arg1) {buffer.update(0, buffer.getInt(0)+1);}/*** 合并 update操作，可能是針對(duì)一個(gè)分組內(nèi)的部分?jǐn)?shù)據(jù)，在某個(gè)節(jié)點(diǎn)上發(fā)生的 但是可能一個(gè)分組內(nèi)的數(shù)據(jù)，會(huì)分布在多個(gè)節(jié)點(diǎn)上處理* 此時(shí)就要用merge操作，將各個(gè)節(jié)點(diǎn)上分布式拼接好的串，合并起來* buffer1.getInt(0) : 大聚和的時(shí)候 上一次聚合后的值 * buffer2.getInt(0) : 這次計(jì)算傳入進(jìn)來的update的結(jié)果* 這里即是：最后在分布式節(jié)點(diǎn)完成后需要進(jìn)行全局級(jí)別的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)));}/*** 初始化一個(gè)內(nèi)部的自己定義的值,在Aggregate之前每組數(shù)據(jù)的初始化結(jié)果*/@Overridepublic void initialize(MutableAggregationBuffer buffer) {buffer.update(0, 0);}/*** 最后返回一個(gè)和DataType的類型要一致的類型，返回UDAF最后的計(jì)算結(jié)果*/@Overridepublic Object evaluate(Row row) {return row.getInt(0);}@Overridepublic boolean deterministic() {//設(shè)置為truereturn true;}/*** 指定UDAF函數(shù)計(jì)算后返回的結(jié)果類型*/@Overridepublic DataType dataType() {return DataTypes.IntegerType;}/*** 在進(jìn)行聚合操作的時(shí)候所要處理的數(shù)據(jù)的結(jié)果的類型*/@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();

開窗函數(shù)

注意：

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

開窗函數(shù)格式：

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

java:

SparkSession sparkSession = SparkSession.builder().appName("window").master("local")//開啟hive的支持，接下來就可以操作hive表了// 前提需要是需要開啟hive metastore 服務(wù).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");/*** 開窗函數(shù)格式：* 【 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();

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。