教育行業(yè)A股IPO第一股(股票代碼 003032)

全國(guó)咨詢/投訴熱線:400-618-4000

Spark SQL快速上手

更新時(shí)間:2015年12月29日15時(shí)58分 來(lái)源:傳智播客云計(jì)算學(xué)科 瀏覽次數(shù):

sparksql結(jié)合hive最佳實(shí)踐
一、Spark SQL快速上手
1、Spark SQL是什么
Spark SQL 是一個(gè)用來(lái)處理結(jié)構(gòu)化數(shù)據(jù)的spark組件。它提供了一個(gè)叫做DataFrames的可編程抽象數(shù)據(jù)模型,并且可被視為一個(gè)分布式的SQL查詢引擎。

2、Spark SQL的基礎(chǔ)數(shù)據(jù)模型-----DataFrames
DataFrame是由“命名列”(類似關(guān)系表的字段定義)所組織起來(lái)的一個(gè)分布式數(shù)據(jù)集合。你可以把它看成是一個(gè)關(guān)系型數(shù)據(jù)庫(kù)的表。
DataFrame可以通過多種來(lái)源創(chuàng)建:結(jié)構(gòu)化數(shù)據(jù)文件,hive的表,外部數(shù)據(jù)庫(kù),或者RDDs

3、Spark SQL如何使用
首先,利用sqlContext從外部數(shù)據(jù)源加載數(shù)據(jù)為DataFrame
然后,利用DataFrame上豐富的api進(jìn)行查詢、轉(zhuǎn)換
最后,將結(jié)果進(jìn)行展現(xiàn)或存儲(chǔ)為各種外部數(shù)據(jù)形式
如圖所示:
 
4、Spark SQL代碼示例
?    加載數(shù)據(jù)
sqlContext支持從各種各樣的數(shù)據(jù)源中創(chuàng)建DataFrame,內(nèi)置支持的數(shù)據(jù)源有parquetFile,jsonFile,外部數(shù)據(jù)庫(kù),hive表,RDD等,另外,hbase等數(shù)據(jù)源的支持也在社區(qū)不斷涌現(xiàn)
# 從Hive中的users表構(gòu)造DataFrame
users = sqlContext.table("users")
# 加載S3上的JSON文件
logs = sqlContext.load("s3n://path/to/data.json", "json")
# 加載HDFS上的Parquet文件
clicks = sqlContext.load("hdfs://path/to/data.parquet", "parquet")
# 通過JDBC訪問MySQL
comments = sqlContext.jdbc("jdbc:mysql://localhost/comments", "user")
# 將普通RDD轉(zhuǎn)變?yōu)镈ataFrame
rdd = sparkContext.textFile("article.txt") \
                  .flatMap(_.split(" ")) \
                  .map((_, 1)) \
                  .reduceByKey(_+_) \
wordCounts = sqlContext.createDataFrame(rdd, ["word", "count"])
 
# 將本地?cái)?shù)據(jù)容器轉(zhuǎn)變?yōu)镈ataFrame
data = [("Alice", 21), ("Bob", 24)]
people = sqlContext.createDataFrame(data, ["name", "age"])

?    使用DataFrame
Spark DataFrame提供了一整套用于操縱數(shù)據(jù)的DSL。這些DSL在語(yǔ)義上與SQL關(guān)系查詢非常相近(這也是Spark SQL能夠?yàn)镈ataFrame提供無(wú)縫支持的重要原因之一)。以下是一組用戶數(shù)據(jù)分析示例:
# 創(chuàng)建一個(gè)只包含年齡小于21歲用戶的DataFrame
young = users.filter(users.age < 21)
 
# 也可以使用Pandas風(fēng)格的語(yǔ)法
young = users[users.age < 21]
# 將所有人的年齡加1
young.select(young.name, young.age + 1)
# 統(tǒng)計(jì)年輕用戶中各性別人數(shù)
young.groupBy("gender").count()
# 將所有年輕用戶與另一個(gè)名為logs的DataFrame聯(lián)接起來(lái)
young.join(logs, logs.userId == users.userId, "left_outer")
除DSL以外,我們當(dāng)然也可以使用熟悉的SQL來(lái)處理DataFrame:
young.registerTempTable("young")
sqlContext.sql("SELECT count(*) FROM young")

?    保存結(jié)果
對(duì)數(shù)據(jù)的分析完成之后,可以將結(jié)果保存在多種形式的外部存儲(chǔ)中
# 追加至HDFS上的Parquet文件
young.save(path="hdfs://path/to/data.parquet", source="parquet", mode="append")
 
# 覆寫S3上的JSON文件
young.save(path="s3n://path/to/data.json", source="json",mode="append")
 
# 保存為Hive的內(nèi)部表
young.saveAsTable(tableName="young", source="parquet" mode="overwrite")
 
# 轉(zhuǎn)換為Pandas DataFrame(Python API特有功能)
pandasDF = young.toPandas()
 
# 以表格形式打印輸出
young.show()


二、SparkSQL操作Hive中的表數(shù)據(jù)
spark可以通過讀取hive的元數(shù)據(jù)來(lái)兼容hive,讀取hive的表數(shù)據(jù),然后在spark引擎中進(jìn)行sql統(tǒng)計(jì)分析,從而,通過sparksql與hive結(jié)合實(shí)現(xiàn)數(shù)據(jù)分析將成為一種最佳實(shí)踐。詳細(xì)實(shí)現(xiàn)步驟如下:

1、啟動(dòng)hive的元數(shù)據(jù)服務(wù)
hive可以通過服務(wù)的形式對(duì)外提供元數(shù)據(jù)讀寫操作,通過簡(jiǎn)單的配置即可
?    編輯 $HIVE_HOME/conf/hive-site.xml,增加如下內(nèi)容:
<property>
<name>hive.metastore.uris</name>
<value>thrift:// hdp-node-01:9083</value>
</property>

?    啟動(dòng)hive metastore
[hadoop@hdp-node-01 ~]${HIVE_HOME}/bin/hive --service metastore  1>/dev/null  2>&1  &

?    查看 metastore:
[hadoop@hdp-node-01 ~] jobs
[1]+ Running hive --service metastore &


2、spark配置
?    將hive的配置文件拷貝給spark
將 $HIVE_HOME/conf/hive-site.xml copy或者軟鏈 到 $SPARK_HOME/conf/

?    將mysql的jdbc驅(qū)動(dòng)包拷貝給spark
將 $HIVE_HOME/lib/mysql-connector-java-5.1.12.jar copy或者軟鏈到$SPARK_HOME/lib/

3、啟動(dòng)spark-sql的shell交互界面
spark-sql已經(jīng)集成在spark-shell中,因此,只要啟動(dòng)spark-shell,就可以使用spakr-sql的shell交互接口:
[hadoop@hdp-node-01 spark] bin/spark-shell --master spark://hdp-node-01:7077
 
或者,可以啟動(dòng)spark-sql界面,使用起來(lái)更方便
[hadoop@hdp-node-01 spark] bin/spark-sql --master spark://hdp-node-01:7077

4、在交互界面輸入sql進(jìn)行查詢
注:以下所用到的庫(kù)和表,都是已經(jīng)在hive中存在的庫(kù)和表

?    如果在spark-shell中執(zhí)行sql查詢,使用sqlContext對(duì)象調(diào)用sql()方法
scala> sqlContext.sql("select remote_addr from dw_weblog.t_ods_detail group by remote_addr").collect.foreach(println)

?    如果是在spark-sql中執(zhí)行sql查詢,則可以直接輸入sql語(yǔ)句
scala> show databases
scala> use dw_weblog
scala> select remote_addr from dw_weblog.t_ods_detail group by remote_addr



5、在IDEA中編寫代碼使用hive-sql
如下所示:
val hiveContext = new HiveContext(sc)
    import hiveContext.implicits._
    import hiveContext.sql
    //指定庫(kù)
sql("use dw_weblog")
//執(zhí)行標(biāo)準(zhǔn)sql語(yǔ)句
sql("create table sparksql as select remote_addr,count(*) from t_ods_detail group by remote_addr")
……




綜上所述,sparksql類似于hive,可以支持sql語(yǔ)法來(lái)對(duì)海量數(shù)據(jù)進(jìn)行分析查詢,跟hive不同的是,hive執(zhí)行sql任務(wù)的底層運(yùn)算引擎采用mapreduce運(yùn)算框架,而sparksql執(zhí)行sql任務(wù)的運(yùn)算引擎是spark core,從而充分利用spark內(nèi)存計(jì)算及DAG模型的優(yōu)勢(shì),大幅提升海量數(shù)據(jù)的分析查詢速度
源碼
sparksql結(jié)合hive最佳實(shí)踐<br />
一、Spark SQL快速上手<br />
1、Spark SQL是什么<br />
Spark SQL 是一個(gè)用來(lái)處理結(jié)構(gòu)化數(shù)據(jù)的spark組件。它提供了一個(gè)叫做DataFrames的可編程抽象數(shù)據(jù)模型,并且可被視為一個(gè)分布式的SQL查詢引擎。<br />
<br />
2、Spark SQL的基礎(chǔ)數(shù)據(jù)模型-----DataFrames<br />
DataFrame是由&ldquo;命名列&rdquo;(類似關(guān)系表的字段定義)所組織起來(lái)的一個(gè)分布式數(shù)據(jù)集合。你可以把它看成是一個(gè)關(guān)系型數(shù)據(jù)庫(kù)的表。<br />
DataFrame可以通過多種來(lái)源創(chuàng)建:結(jié)構(gòu)化數(shù)據(jù)文件,hive的表,外部數(shù)據(jù)庫(kù),或者RDDs<br />
<br />
3、Spark SQL如何使用<br />
首先,利用sqlContext從外部數(shù)據(jù)源加載數(shù)據(jù)為DataFrame<br />
然后,利用DataFrame上豐富的api進(jìn)行查詢、轉(zhuǎn)換<br />
最后,將結(jié)果進(jìn)行展現(xiàn)或存儲(chǔ)為各種外部數(shù)據(jù)形式<br />
如圖所示:
<div style="text-align: center;"><img alt="" src="/files/image/201512/20151229153449294.jpg" style="width: 400px; height: 253px;" /></div>
&nbsp;<br />
4、Spark SQL代碼示例<br />
?&nbsp;&nbsp; &nbsp;加載數(shù)據(jù)<br />
sqlContext支持從各種各樣的數(shù)據(jù)源中創(chuàng)建DataFrame,內(nèi)置支持的數(shù)據(jù)源有parquetFile,jsonFile,外部數(shù)據(jù)庫(kù),hive表,RDD等,另外,hbase等數(shù)據(jù)源的支持也在社區(qū)不斷涌現(xiàn)<br />
# 從Hive中的users表構(gòu)造DataFrame<br />
users = sqlContext.table(&quot;users&quot;)<br />
# 加載S3上的JSON文件<br />
logs = sqlContext.load(&quot;s3n://path/to/data.json&quot;, &quot;json&quot;)<br />
# 加載HDFS上的Parquet文件<br />
clicks = sqlContext.load(&quot;hdfs://path/to/data.parquet&quot;, &quot;parquet&quot;)<br />
# 通過JDBC訪問MySQL<br />
comments = sqlContext.jdbc(&quot;jdbc:mysql://localhost/comments&quot;, &quot;user&quot;)<br />
# 將普通RDD轉(zhuǎn)變?yōu)镈ataFrame<br />
rdd = sparkContext.textFile(&quot;article.txt&quot;) \<br />
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .flatMap(_.split(&quot; &quot;)) \<br />
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .map((_, 1)) \<br />
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .reduceByKey(_+_) \<br />
wordCounts = sqlContext.createDataFrame(rdd, [&quot;word&quot;, &quot;count&quot;])<br />
&nbsp;<br />
# 將本地?cái)?shù)據(jù)容器轉(zhuǎn)變?yōu)镈ataFrame<br />
data = [(&quot;Alice&quot;, 21), (&quot;Bob&quot;, 24)]<br />
people = sqlContext.createDataFrame(data, [&quot;name&quot;, &quot;age&quot;])<br />
<br />
?&nbsp;&nbsp; &nbsp;使用DataFrame<br />
Spark DataFrame提供了一整套用于操縱數(shù)據(jù)的DSL。這些DSL在語(yǔ)義上與SQL關(guān)系查詢非常相近(這也是Spark SQL能夠?yàn)镈ataFrame提供無(wú)縫支持的重要原因之一)。以下是一組用戶數(shù)據(jù)分析示例:<br />
# 創(chuàng)建一個(gè)只包含年齡小于21歲用戶的DataFrame<br />
young = users.filter(users.age &lt; 21)<br />
&nbsp;<br />
# 也可以使用Pandas風(fēng)格的語(yǔ)法<br />
young = users[users.age &lt; 21]<br />
# 將所有人的年齡加1<br />
young.select(young.name, young.age + 1)<br />
# 統(tǒng)計(jì)年輕用戶中各性別人數(shù)<br />
young.groupBy(&quot;gender&quot;).count()<br />
# 將所有年輕用戶與另一個(gè)名為logs的DataFrame聯(lián)接起來(lái)<br />
young.join(logs, logs.userId == users.userId, &quot;left_outer&quot;)<br />
除DSL以外,我們當(dāng)然也可以使用熟悉的SQL來(lái)處理DataFrame:<br />
young.registerTempTable(&quot;young&quot;)<br />
sqlContext.sql(&quot;SELECT count(*) FROM young&quot;)<br />
<br />
?&nbsp;&nbsp; &nbsp;保存結(jié)果<br />
對(duì)數(shù)據(jù)的分析完成之后,可以將結(jié)果保存在多種形式的外部存儲(chǔ)中<br />
# 追加至HDFS上的Parquet文件<br />
young.save(path=&quot;hdfs://path/to/data.parquet&quot;, source=&quot;parquet&quot;, mode=&quot;append&quot;)<br />
&nbsp;<br />
# 覆寫S3上的JSON文件<br />
young.save(path=&quot;s3n://path/to/data.json&quot;, source=&quot;json&quot;,mode=&quot;append&quot;)<br />
&nbsp;<br />
# 保存為Hive的內(nèi)部表<br />
young.saveAsTable(tableName=&quot;young&quot;, source=&quot;parquet&quot; mode=&quot;overwrite&quot;)<br />
&nbsp;<br />
# 轉(zhuǎn)換為Pandas DataFrame(Python API特有功能)<br />
pandasDF = young.toPandas()<br />
&nbsp;<br />
# 以表格形式打印輸出<br />
young.show()<br />
<br />
<br />
二、SparkSQL操作Hive中的表數(shù)據(jù)<br />
spark可以通過讀取hive的元數(shù)據(jù)來(lái)兼容hive,讀取hive的表數(shù)據(jù),然后在spark引擎中進(jìn)行sql統(tǒng)計(jì)分析,從而,通過sparksql與hive結(jié)合實(shí)現(xiàn)數(shù)據(jù)分析將成為一種最佳實(shí)踐。詳細(xì)實(shí)現(xiàn)步驟如下:<br />
<br />
1、啟動(dòng)hive的元數(shù)據(jù)服務(wù)<br />
hive可以通過服務(wù)的形式對(duì)外提供元數(shù)據(jù)讀寫操作,通過簡(jiǎn)單的配置即可<br />
?&nbsp;&nbsp; &nbsp;編輯 $HIVE_HOME/conf/hive-site.xml,增加如下內(nèi)容:<br />
&lt;property&gt;<br />
&lt;name&gt;hive.metastore.uris&lt;/name&gt;<br />
&lt;value&gt;thrift:// hdp-node-01:9083&lt;/value&gt;<br />
&lt;/property&gt;<br />
<br />
?&nbsp;&nbsp; &nbsp;啟動(dòng)hive metastore<br />
[hadoop@hdp-node-01 ~]${HIVE_HOME}/bin/hive --service metastore&nbsp; 1&gt;/dev/null&nbsp; 2&gt;&amp;1&nbsp; &amp;<br />
<br />
?&nbsp;&nbsp; &nbsp;查看 metastore:<br />
[hadoop@hdp-node-01 ~] jobs<br />
[1]+ Running hive --service metastore &amp;<br />
<br />
<br />
2、spark配置<br />
?&nbsp;&nbsp; &nbsp;將hive的配置文件拷貝給spark<br />
將 $HIVE_HOME/conf/hive-site.xml copy或者軟鏈 到 $SPARK_HOME/conf/<br />
<br />
?&nbsp;&nbsp; &nbsp;將mysql的jdbc驅(qū)動(dòng)包拷貝給spark<br />
將 $HIVE_HOME/lib/mysql-connector-java-5.1.12.jar copy或者軟鏈到$SPARK_HOME/lib/<br />
<br />
3、啟動(dòng)spark-sql的shell交互界面<br />
spark-sql已經(jīng)集成在spark-shell中,因此,只要啟動(dòng)spark-shell,就可以使用spakr-sql的shell交互接口:<br />
[hadoop@hdp-node-01 spark] bin/spark-shell --master spark://hdp-node-01:7077
<div style="text-align: center;"><img alt="" src="/files/image/201512/20151229153518583.png" style="width: 400px; height: 170px;" />&nbsp;</div>
<br />
或者,可以啟動(dòng)spark-sql界面,使用起來(lái)更方便<br />
[hadoop@hdp-node-01 spark] bin/spark-sql --master spark://hdp-node-01:7077<br />
<br />
4、在交互界面輸入sql進(jìn)行查詢<br />
注:以下所用到的庫(kù)和表,都是已經(jīng)在hive中存在的庫(kù)和表<br />
<br />
?&nbsp;&nbsp; &nbsp;如果在spark-shell中執(zhí)行sql查詢,使用sqlContext對(duì)象調(diào)用sql()方法<br />
scala&gt; sqlContext.sql(&quot;select remote_addr from dw_weblog.t_ods_detail group by remote_addr&quot;).collect.foreach(println)<br />
<br />
?&nbsp;&nbsp; &nbsp;如果是在spark-sql中執(zhí)行sql查詢,則可以直接輸入sql語(yǔ)句<br />
scala&gt; show databases<br />
scala&gt; use dw_weblog<br />
scala&gt; select remote_addr from dw_weblog.t_ods_detail group by remote_addr<br />
<br />
<br />
<br />
5、在IDEA中編寫代碼使用hive-sql<br />
如下所示:<br />
val hiveContext = new HiveContext(sc)<br />
&nbsp;&nbsp;&nbsp; import hiveContext.implicits._<br />
&nbsp;&nbsp;&nbsp; import hiveContext.sql<br />
&nbsp;&nbsp;&nbsp; //指定庫(kù)<br />
sql(&quot;use dw_weblog&quot;)<br />
//執(zhí)行標(biāo)準(zhǔn)sql語(yǔ)句<br />
sql(&quot;create table sparksql as select remote_addr,count(*) from t_ods_detail group by remote_addr&quot;)<br />
&hellip;&hellip;<br />
<br />
<br />
<br />
<br />
綜上所述,sparksql類似于hive,可以支持sql語(yǔ)法來(lái)對(duì)海量數(shù)據(jù)進(jìn)行分析查詢,跟hive不同的是,hive執(zhí)行sql任務(wù)的底層運(yùn)算引擎采用mapreduce運(yùn)算框架,而sparksql執(zhí)行sql任務(wù)的運(yùn)算引擎是spark core,從而充分利用spark內(nèi)存計(jì)算及DAG模型的優(yōu)勢(shì),大幅提升海量數(shù)據(jù)的分析查詢速度<br />
0 分享到:
和我們?cè)诰€交談!