Spark入门（第1讲）
一、Spark是什么
引用官方文档的一句话
Apache Spark™ is a unified analytics engine for large-scale data processing.
Apache Spark™是用于大规模数据处理的统一分析引擎。
可以从这句话拆分出几个关键点
- 统一
- 大数据
- 分析引擎/计算引擎
何为统一
Spark的主要目标是为编写大数据应用程序提供统一的平台，其中包括统一计算引擎和统一API。
Spark提供一致的，可组合的API来构建应用程序，使得任务的编写更加高效。
例如：使用SQL加载数据，然后使用Spark的ML库评估其上的机器学习模型，引擎能够将这些步骤合并成一次数据扫描。
通用API设计+高性能执行
同时Spark为了实现统一，不仅支持使用自带的标准库，为通用数据分析人物提供统一的API，也支持由开源社区以第三方包形式发布的大量外部库。
何为计算引擎
不同于其他大数据软件平台(例如Hadoop)，Spark仅负责加载数据并计算，并不负责数据永久存储(持久化)。因此，Spark可以与多种持久化存储系统结合使用。常用的有以下几种
- 云存储系统（Azure存储、Amazon S3）
- 分布式文件系统（Apache Hadoop）
- 键值存储系统（Apache Cassandra）
- 消息队列系统（Apache Kafka）
二、Spark基本架构
Spark应用程序由一个驱动器进程和一组执行器进程组成。
驱动进程负责：
- 维护Spark应用程序的相关信息
- 回应用户的程序或输入
- 分析任务并分发给若干执行器处理
驱动器负责：
- 执行驱动器分配给他的代码
- 汇报执行器计算状态给驱动器
这种分发可以交给Spark的集群管理器来处理，在一个任务提交给集群管理器后，集群管理器会将计算资源分配给应用程序。集群管理器可以是三个核心集群管理器之一：Spark独立集群管理器，Yarn、Mesos
三、Spark基本概念
SparkSession
Spark API支持多种语言来启动Spark任务，各种进程通过创建SparkSession的方式将用户命令和数据发送给Spark。
DataFrame
包含行和列的数据表，区别于普通的电子表格，DataFrame支持分布式存储。用于说明这些列和列类型的一些规则称为schema
数据分区
为了让多个执行器并行地工作，Spark将数据分解为多个数据块，每个数据块叫做一个分区
转换操作
将一个抽象的操作指定给一个操作对象，但不会立即执行。有两类转换操作：
-
指定窄依赖关系（narrow dependency）的转换操作
每个输出分区只影响一个输出分区。
在内存中执行多个转换操作
-
指定宽依赖关系（wide dependency）的转换操作 ==》shuffle
每个输出分区决定多个输出分区。
将结果写入磁盘
惰性评估（lazy evaluation）
等到绝对需要时才执行计算。
用户表达对数据的一些操作，不会立刻修改数据，而是建立一个作用到原始数据的转换计划。Spark将计划编译为可以在及群众高效运行的流水线式的物理执行计划。等到最后时刻才执行代码。
动作操作
- 在控制台查看数据的动作
- 在某个语言中将数据汇集为原生对象的动作
- 写入输出数据源的动作
上述概念看着可能比较抽象，用实际例子标识
//创建一个DataFrame
scala> var flightData2015 = spark.read.option("inferSchema","true").option("header","true").csv("/Users/jacobzheng/IdeaProjects/Spark-The-Definitive-Guide/data/flight-data/csv/2015-summary.csv")
//返回
flightData2015: org.apache.spark.sql.DataFrame = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]
//取数据
scala> flightData2015.take(3)
res3: Array[org.apache.spark.sql.Row] = Array([United States,Romania,15], [United States,Croatia,1], [United States,Ireland,344])
//排序后取数据
scala> flightData2015.sort("count").take(3)
res4: Array[org.apache.spark.sql.Row] = Array([Moldova,United States,1], [United States,Croatia,1], [United States,Singapore,1])
//取数据
scala> flightData2015.take(3)
res5: Array[org.apache.spark.sql.Row] = Array([United States,Romania,15], [United States,Croatia,1], [United States,Ireland,344])
从上述例子可以看出
Sort操作不会修改原有的DataFrame，而是生成一个新的DataFrame。
DataFrame和SQL
我理解（仅限于我理解，可能是错的）
DataFrame和SQL都是一种逻辑框架，Spark通过解析这种逻辑编译出一个用于底层执行的执行计划。
DataFrame可以使用简单的函数注册一个表或者视图
flightData2015.createOrReplaceTempView("flightData2015")
通过SparkSql查询，结果将会返回一个DataFrame
spark.sql("select max(count) from flightData2015")
使用DataFrame和SparkSql执行一样的操作，执行计划是一样的。
四、Spark工具集
spark-submit
将测试级别的交互式程序转化为生产级别的应用程序。
spark-submit --class org.apache.spark.examples/SparkPi --master local ../examples/jars/spark-examples_2.11-2.4.6.jar
通过修改master参数，可以将应用程序提交到集群上，集群需要运行Spark standalone集群、Mesos或Yarn。
Dataset
Dataset是一种类型安全的Spark结构化API，与DataFrame不同的是，Dataset api能够让用户用Java/scala定义DataFrame中的每条记录。
//定义一个Dataset
scala> case class Flight(DEST_COUNTRY_NAME: String,ORIGIN_COUNTRY_NAME: String,count: BigInt)
defined class Flight
//读取parquet文件到DataFrame
scala> var flightDF = spark.read.parquet("/Users/jacobzheng/IdeaProjects/Spark-The-Definitive-Guide/data/flight-data/parquet/2010-summary.parquet")
flightDF: org.apache.spark.sql.DataFrame = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]
//将DataFrame转换为Dataset
scala> val flights = flightDF.as[Flight]
flights: org.apache.spark.sql.Dataset[Flight] = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]
结构化流处理
可以减少延迟并允许增量更新，可以快速地从流式系统中提取数据而几乎不需要修改代码。可以按照传统批式处理作业的模式进行设计，然后将其转换为流式作业。
机器学习和高级数据分析
低级API
SparkR
附录
Spark安装实录
0.环境准备
| 系统版本 | MacOS 10.15.4 |
|---|---|
| MacOS 10.15.4 | 1.8.0_111 |
1.在官网上下载Spark安装包
在这里我大胆选择了最新版的Spark+最新版的Hadoop3.2
下载完成后解压即可
ps.我把解压后的目录软链到了~/spark目录下了
2.检查是否可用
#进入Spark目录下
~/spark $ bin/spark-shell
20/06/11 13:58:53 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
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).
Spark context Web UI available at http://192.168.199.36:4040
Spark context available as 'sc' (master = local[*], app id = local-1591855138249).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 3.0.0-preview2
/_/
Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_111)
Type in expressions to have them evaluated.
Type :help for more information.
scala>
3.配置环境变量**
#spark
export SAPRK_HOME=/Users/jacobzheng/spark
export PATH=$PATH:$SPARK_HOME/bin
配置完成后source一下使得环境变量生效
然后直接spark-shell会有和步骤2一样的效果
4.启动master
~/spark $ ./sbin/start-master.sh
starting org.apache.spark.deploy.master.Master, logging to /Users/jacobzheng/spark/logs/spark-jacobzheng-org.apache.spark.deploy.master.Master-1-MacBook-Pro.local.out
根据输出提示的目录地址，查看spark服务日志
~/spark $ tail -500f /Users/jacobzheng/spark/logs/spark-jacobzheng-org.apache.spark.deploy.master.Master-1-MacBook-Pro.local.out
Spark Command: /Library/Java/JavaVirtualMachines/jdk1.8.0_111.jdk/Contents/Home//bin/java -cp /Users/jacobzheng/spark/conf/:/Users/jacobzheng/spark/jars/* -Xmx1g org.apache.spark.deploy.master.Master --host MacBook-Pro.local --port 7077 --webui-port 8080
========================================
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
20/06/11 14:10:53 INFO Master: Started daemon with process name: 74179@MacBook-Pro.local
20/06/11 14:10:53 INFO SignalUtils: Registered signal handler for TERM
20/06/11 14:10:53 INFO SignalUtils: Registered signal handler for HUP
20/06/11 14:10:53 INFO SignalUtils: Registered signal handler for INT
20/06/11 14:10:53 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
20/06/11 14:10:53 INFO SecurityManager: Changing view acls to: jacobzheng
20/06/11 14:10:53 INFO SecurityManager: Changing modify acls to: jacobzheng
20/06/11 14:10:53 INFO SecurityManager: Changing view acls groups to:
20/06/11 14:10:53 INFO SecurityManager: Changing modify acls groups to:
20/06/11 14:10:53 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(jacobzheng); groups with view permissions: Set(); users with modify permissions: Set(jacobzheng); groups with modify permissions: Set()
20/06/11 14:10:53 INFO Utils: Successfully started service 'sparkMaster' on port 7077.
20/06/11 14:10:53 INFO Master: Starting Spark master at spark://MacBook-Pro.local:7077
20/06/11 14:10:54 INFO Master: Running Spark version 3.0.0-preview2
20/06/11 14:10:54 INFO Utils: Successfully started service 'MasterUI' on port 8080.
20/06/11 14:10:54 INFO MasterWebUI: Bound MasterWebUI to 0.0.0.0, and started at http://192.168.199.36:8080
20/06/11 14:10:54 INFO Master: I have been elected leader! New state: ALIVE
可以看到sparkMaster服务监听了7077端口，MasterUI监听了8080端口
访问 http://localhost:8080/ 来查看到当前master的总体状态。
5.启动Worker
~/spark $ spark-class org.apache.spark.deploy.worker.Worker spark://MacBook-Pro.local:7077
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
20/06/11 14:17:13 INFO Worker: Started daemon with process name: 74367@MacBook-Pro.local
20/06/11 14:17:13 INFO SignalUtils: Registered signal handler for TERM
20/06/11 14:17:13 INFO SignalUtils: Registered signal handler for HUP
20/06/11 14:17:13 INFO SignalUtils: Registered signal handler for INT
20/06/11 14:17:13 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
20/06/11 14:17:13 INFO SecurityManager: Changing view acls to: jacobzheng
20/06/11 14:17:13 INFO SecurityManager: Changing modify acls to: jacobzheng
20/06/11 14:17:13 INFO SecurityManager: Changing view acls groups to:
20/06/11 14:17:13 INFO SecurityManager: Changing modify acls groups to:
20/06/11 14:17:13 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(jacobzheng); groups with view permissions: Set(); users with modify permissions: Set(jacobzheng); groups with modify permissions: Set()
20/06/11 14:17:13 INFO Utils: Successfully started service 'sparkWorker' on port 55878.
20/06/11 14:17:13 INFO Worker: Starting Spark worker 192.168.199.36:55878 with 16 cores, 15.0 GiB RAM
20/06/11 14:17:14 INFO Worker: Running Spark version 3.0.0-preview2
20/06/11 14:17:14 INFO Worker: Spark home: /Users/jacobzheng/spark
20/06/11 14:17:14 INFO ResourceUtils: ==============================================================
20/06/11 14:17:14 INFO ResourceUtils: Resources for spark.worker:
20/06/11 14:17:14 INFO ResourceUtils: ==============================================================
20/06/11 14:17:14 INFO Utils: Successfully started service 'WorkerUI' on port 8081.
20/06/11 14:17:14 INFO WorkerWebUI: Bound WorkerWebUI to 0.0.0.0, and started at http://192.168.199.36:8081
20/06/11 14:17:14 INFO Worker: Connecting to master MacBook-Pro.local:7077...
20/06/11 14:17:14 INFO TransportClientFactory: Successfully created connection to MacBook-Pro.local/192.168.199.36:7077 after 41 ms (0 ms spent in bootstraps)
20/06/11 14:17:14 INFO Worker: Successfully registered with master spark://MacBook-Pro.local:7077
可以看到启动了一个sparkWorker监听55878，启动了WorkerUI监听8081端口
此时刷新masterUI可以看到多了一个worker的信息
点击进去可以看到worker的信息