【數據倉庫】spark大數據處理框架

文章目錄

- 概述
- 架構
- spark 架構角色
- 下載
- 安裝
- 啟動pyspark
- 啟動spark-sehll
- 啟動spark-sql
- spark-submit
- 經驗

概述

Spark是一個性能優異的集群計算框架，廣泛應用于大數據領域。類似Hadoop，但對Hadoop做了優化，計算任務的中間結果可以存儲在內存中，不需要每次都寫入HDFS，更適用于需要迭代運算的算法場景中。

Spark專注于數據的處理分析，而數據的存儲還是要借助于Hadoop分布式文件系統HDFS等來實現。

大數據問題場景包含以下三種：

復雜的批量數據處理
基于歷史數據的交互式查詢
基于實時數據流的數據處理

Spark技術棧基本可以解決以上三種場景問題。

架構

在這里插入圖片描述

1 spark Core :spark的核心模塊，是spark運行基礎。以RDD為數據抽象，提供python、java、scala、R語言的api,可以通過RDD編程進行海量離線數據批處理計算。

2 Spark SQL：基于Spark Core，提供結構化數據處理功能。可以使用SQL語言對數據進行處理，可用于離線計算場景。同時基于Spark SQL提供了StructuredStreaming模塊，可以使用時SQL進行流式計算。

3 sparkStreaming : 以Spark Core為基礎，提供數據的流式計算功能

4 MLlib:以spark Core為基礎，進行機器學習計算，內置大量機器學習庫和API算法等。

5 Graphx：以spark Core為基礎，進行圖計算，提供大量圖計算的API，方便以分布式資源進行圖計算。

6 spark底層的文件存儲還是基于hdfs分布式文件系統，支持多種部署方式。

spark 架構角色

在這里插入圖片描述

從兩個層面理解：

資源管理層面：（典型的Master-Worker架構）

管理者：即Master角色，只能有一個

? 工作者：即Worker角色，可以有多個。一個worker在一個分布式節點上，監測當前節點的資源狀況，向master節點匯總。

任務執行層面：

? 某任務管理者：Driver角色，一個任務只能有一個

? 某任務執行者：Executor角色，可以有多個

在特殊場景下（local模式）,Driver即是管理者又是執行者

下載

下載地址：

http://spark.apache.org/downloads.html

或者

https://archive.apache.org/dist/spark/

選擇合適自己的版本下載。

Spark2.X預編譯了Scala2.11(Spark2.4.2預編譯Scala2.12)

Spark3.0+預編譯了Scala2.12

該教程選擇Spark3.2.1版本，其中預編譯了Hadoop3.2和Scala2.13，對應的包是 spark-3.2.1-bin-hadoop3.2-scala2.13.tgz，但這里的預編譯Hadoop不是指不需要再安裝Hadoop。

linux 服務器上下載地址

wget https://archive.apache.org/dist/spark/spark-3.2.1/spark-3.2.1-bin-hadoop3.2-scala2.13.tgz

安裝

Spark的安裝部署支持三種模式，

local本地模式（單機）：啟動一個JVM Process進程，通過其內部的多個線程來模擬整個spark運行時各個角色。一個進程里有多個線程。

Local[N]:可以使用N個線程，一個線程利用一個cpu核，通常cpu有幾個核，就指定幾個線程，最大化利用計算能力；
Local[*]，按照cpu核數設置線程數；

standalone模式（集群）：各個角色以獨立進程的形式存在，并組成spark集群

? spark on YARN模式（集群）：各個角色運行在yarn的容器內部，組成集群環境

? kubernetes 模式（容器集群）：各個角色運行在kubernetes 容器內部，組成集群環境

本文將只介紹本地Local模式，其它模式將會在后續文章中進行介紹。

該文的安裝環境為centos7。

1、將下載的包上傳到服務器指定目錄，解壓

[root@localhost softnew]# tar zxvf spark-3.1.2-bin-hadoop3.2.tgz
# 修改目錄
mv spark-3.1.2-bin-hadoop3.2 spark-3.1.2

2、修改配置文件

修改/etc/profile文件，新增spark環境變量：

 # Spark Environment Variablesexport SPARK_HOME=/home/bigData/softnew/sparkexport PATH=$PATH:$SPARK_HOME/bin

修改完成后記得執行 source /etc/profile 使其生效

啟動pyspark

? pyspark 是spark集成python后，可以使用python 腳本編寫spark 數據批處理計算。pyspark提供了一個shell窗口。

./pyspark

[root@yd-ss bin]# ./pyspark
Python 3.10.10 (main, Dec 26 2024, 22:46:13) [GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
Type "help", "copyright", "credits" or "license" for more information.
24/12/27 10:46:44 WARN Utils: Your hostname, yd-ss resolves to a loopback address: 127.0.0.1; using xx.xx.xx.xx instead (on interface bond0)
24/12/27 10:46:44 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
24/12/27 10:46:46 WARN HiveConf: HiveConf of name hive.metastore.event.db.notification.api.auth does not exist
24/12/27 10:46:46 WARN HiveConf: HiveConf of name hive.server2.active.passive.ha.enable does not exist
24/12/27 10:46:46 WARN HiveConf: HiveConf of name hive.exec.default.charset does not exist
24/12/27 10:46:46 WARN HiveConf: HiveConf of name hive.exec.default.national.charset does not exist
24/12/27 10:46:46 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Welcome to____              __/ __/__  ___ _____/ /___\ \/ _ \/ _ `/ __/  '_//__ / .__/\_,_/_/ /_/\_\   version 3.2.1/_/Using Python version 3.10.10 (main, Dec 26 2024 22:46:13)
Spark context Web UI available at http://sc:4040
Spark context available as 'sc' (master = local[*], app id = local-1735267609271).
SparkSession available as 'spark'.
>>>

進入窗口，即可使用python 寫RDD編程代碼了。

同時，可以通過web ui 在4040端口訪問，查看spark 任務執行情況。

執行如下計算任務

sc.parallelize([1,2,3,4,5]).map(lambda x:x*10).collect()

訪問localhost:4040

在這里插入圖片描述

可以看到job清單，這個job，起了24個線程去處理計算。由于跑任務的服務器是24核的，執行./pyspark 默認以local[*]最大線程去啟動。

在這里插入圖片描述

可以看到任務層面，啟動了一個driver,由于是local模式，所以driver即是管理者也是執行者。

可以在pyspark-shell下利用spark做一些簡單開發任務；

下面修改啟動命令：

# 該local模式啟動2個線程
./pyspark --master local[2]

再次執行

sc.parallelize([1,2,3,4,5]).map(lambda x:x*10).collect()

在這里插入圖片描述

可以看到這個job只用了2個線程來處理計算。

還可以利用該shell處理其他計算任務，也就是說一個shell 啟動起來，是可以處理多個任務的，但只要關閉窗口，shell就會關閉。就不能再處理任務了。

通過shell 總是不便，后續將介紹通過pycharm進行RDD計算任務編寫。

退出shell腳本

quit()或者ctrl + D

啟動spark-sehll

./spark-shell

可以看到如下信息：

[root@yd-ss bin]# ./spark-shell
24/12/27 11:11:50 WARN Utils: Your hostname, yd-ss resolves to a loopback address: 127.0.0.1; using xx.xx.xx.xx instead (on interface bond0)
24/12/27 11:11:50 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Welcome to____              __/ __/__  ___ _____/ /___\ \/ _ \/ _ `/ __/  '_//___/ .__/\_,_/_/ /_/\_\   version 3.2.1/_/Using Scala version 2.13.5 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201)
Type in expressions to have them evaluated.
Type :help for more information.
24/12/27 11:12:04 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://sc:4040
Spark context available as 'sc' (master = local[*], app id = local-1735269126553).
Spark session available as 'spark'.scala>

這個是要使用scala語言編寫，其他跟pyspark類似。

啟動spark-sql

./spark-sql

可以看到如下：

[root@yd-ss bin]# ./spark-sql
24/12/27 11:14:28 WARN Utils: Your hostname, yd-ss resolves to a loopback address: 127.0.0.1; using xx.xx.xx.xx instead (on interface bond0)
24/12/27 11:14:28 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
24/12/27 11:14:29 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
24/12/27 11:14:30 WARN HiveConf: HiveConf of name hive.metastore.event.db.notification.api.auth does not exist
24/12/27 11:14:30 WARN HiveConf: HiveConf of name hive.server2.active.passive.ha.enable does not exist
24/12/27 11:14:30 WARN HiveConf: HiveConf of name hive.exec.default.charset does not exist
24/12/27 11:14:30 WARN HiveConf: HiveConf of name hive.exec.default.national.charset does not exist
Spark master: local[*], Application Id: local-1735269273943
spark-sql>

可以看到這個是依賴hive數倉配置的。spark-sql是沒有元數據管理的，所以需要跟hive集成，利用其元數據管理功能。后續將詳細介紹。

spark-submit

該工具是用來提交寫好的計算腳本，到saprk上去執行，執行完成即結束。和前面的shell不一樣，shell只要沒關閉，就可以一直執行的。

# 執行spark自帶的python示例，計算pi的值（8次迭代）
./spark-submit /home/spark/spark-3.2.1/examples/src/main/python/pi.py 8

該腳本，會基于spark啟動一個driver,執行pi.py計算，然后打開web ui 4040監控接口，執行完成后輸出結果，最后關閉driver，關閉web ui。

是個一次性的任務執行。

經驗

1 spark 功能比較強大，使用方式也很豐富，初步學習只需要了解自己使用方式即可；

2 spark local模式使用配置是比較簡單的，基本是開箱即用；