Big Data 之 Spark

什么是Spark

官方网站给出的定义是:
Apache Spark™ is a unified analytics engine for large-scale data processing (是一个用于大规模数据处理提供的统一的数据分析引擎)

Spark的历史

1.2009年，Spark诞生于美国加州大学伯克利分校 (UC Berkeley) 的AMP (Algorithms, Machines and People) 实验室，项目采用Scala语言编写
2.2010年，Spark 正式对外开源
3.2013年，Spark加入Apache孵化项目
4.如今，成为Apache的最重要的三大分布式计算系统开源项目之一（Hadoop、Spark、Storm）
5.当前版本已经更新到3.0.0 (截止Jun 18, 2020)

Spark的主要特点

运行快速 (Speed): Spark使用先进的DAG (Directed Acyclic Graph, 有向无环图)执行引擎，以支持循环数据流与内存计算，基于内存的执行速度可比Hadoop MapReduce快上百倍，基于磁盘的执行速度也能快十倍；

Logistic regression in Hadoop and Spark

易用性 (Ease of Use): Spark支持使用Scala、Java、Python和R语言进行编程，简洁的API设计有助于用户轻松构建并行程序，并且可以通过Spark Shell进行交互式编程；

Spark's Python DataFrame API Read JSON files with automatic schema inference

通用性 (Generality): Spark提供了完整而强大的技术栈，包括SQL查询、流式计算、机器学习和图算法组件，这些组件可以无缝整合在同一个应用中，足以应对复杂的计算；

运行模式多样(Runs Everywhere)：Spark可运行于独立的集群模式中，或者运行于Hadoop中，也可运行于Amazon EC2等云环境中，并且可以访问HDFS、Cassandra、HBase、Hive等多种数据源

Spark生态圈

spark的数据源支持hdfs、hbase, 还包括s3、cassdra、mongoDB、mysql等

Spark的一些基本概念

• RDD：是弹性分布式数据集（Resilient Distributed Dataset）的简称，是分布式内存的一个抽象概念，提供了一种高度受限的共享内存模型；
• DAG：是Directed Acyclic Graph（有向无环图）的简称，反映RDD之间的依赖关系；
• Executor：是运行在工作节点（Worker Node）上的一个进程，负责运行任务，并为应用程序存储数据；
• 应用：用户编写的Spark应用程序；
• 任务：运行在Executor上的工作单元；
• 作业：一个作业包含多个RDD及作用于相应RDD上的各种操作；
• 阶段：是作业的基本调度单位，一个作业会分为多组任务，每组任务被称为“阶段”，或者也被称为“任务集”。

架构设计

如下图所示，Spark运行架构包括集群资源管理器（Cluster Manager）、运行作业任务的工作节点（Worker Node）、每个应用的任务控制节点（Driver）和每个工作节点上负责具体任务的执行进程（Executor）。其中，集群资源管理器可以是Spark自带的资源管理器，也可以是YARN或Mesos等资源管理框架。

与Hadoop MapReduce计算框架相比，Spark所采用的Executor有两个优点：一是利用多线程来执行具体的任务（Hadoop MapReduce采用的是进程模型），减少任务的启动开销；二是Executor中有一个BlockManager存储模块，会将内存和磁盘共同作为存储设备，当需要多轮迭代计算时，可以将中间结果存储到这个存储模块里，下次需要时，就可以直接读该存储模块里的数据，而不需要读写到HDFS等文件系统里，因而有效减少了IO开销；或者在交互式查询场景下，预先将表缓存到该存储系统上，从而可以提高读写IO性能

Spark的安装

1.下载和解压Spark安装包
从Spark官网(https://spark.apache.org/downloads.html)下载安装包,选择Pre-built for Apache Hadoop 2.7,下载对应安装包spark-3.0.0-bin-hadoop2.7.tgz，并解压到指定安装目录:
tar -zxvf sparl-3.0.0-bin-hadoop2.7.tgz -C /usr/local
然后重命名为spark-local
cp -r spark-3.0.0-bin-hadoop2.7 spark-local

2.设置Spark的环境变量以启动PySpark,将以下内容添加到~/.bashrc, 然后source .bashrc

export SPARK_HOME = /usr/local/spark-local
export PATH = $PATH:/usr/local/spark-local/bin
export PYTHONPATH = $SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH
export PATH = $SPARK_HOME/python:$PATH

3.启动PySpark shell,就可以在shell上面进行一些交互式的测试
./bin/pyspark

Python 3.7.0 (default, July 19 2020, 06:48:10)
[GCC 5.4.0 20200719] on linux2
Type "help", "copyright", "credits" or "license" for more information.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.0.0
      /_/
Using Python version 3.7.0 (default, July 19 2020 06:48:10)
SparkSession available as 'spark'.
<<<

4.PySpark命令及其常用参数
./bin/pyspark --master <master-url>
Spark的运行模式取决于传递给SparkContext的Master URL的值。Master URL可以是以下任一种式：
* local 使用一个Worker线程本地化运行SPARK(完全不并行)
* local[*] 使用逻辑CPU个数数量的线程来本地化运行Spark
* local[K] 使用K个Worker线程本地化运行Spark（理想情况下，K应该根据运行机器的CPU核数设定）
* spark://HOST:PORT 连接到指定的Spark standalone master。默认端口是7077.
* yarn-client 以客户端模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找 到。
* yarn-cluster 以集群模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找到。
* mesos://HOST:PORT 连接到指定的Mesos集群。默认接口是5050。
需要强调的是，这里我们采用“本地模式”（local）运行Spark,系统默认是“bin/pyspark-master local[*]”,也就是使用本地所有CPU

Spark独立应用编程

接着我们通过一个简单的应用程序来演示如何通过 Spark API 编写一个独立应用程序。使用 Python进行spark编程比Java和Scala简单得多。

在进行Python编程前，请先确定是否已经.bashrc中添加PYTHONPATH环境变量。
接下来即可进行Python编程.

新建一个test.py文件,并在test.py添加代码

from pyspark import SparkContext
sc = SparkContext( 'local', 'test')
logFile = "file:///usr/local/spark/README.md"
logData = sc.textFile(logFile, 2).cache()
numAs = logData.filter(lambda line: 'a' in line).count()
numBs = logData.filter(lambda line: 'b' in line).count()
print('Lines with a: %s, Lines with b: %s' % (numAs, numBs))

$ python3 test.py
>>> Lines with a: 61, Lines with b: 30