Spark2.1.0——Spark初体验

      学习一个工具的最好途径，就是使用它。这就比如《极品飞车》玩得好的同窗，未必真的会开车，要学习车的驾驶技能，就必须用手触摸方向盘、用脚感觉刹车与油门的力道。在IT领域，在深刻了解一个系统的原理、实现细节以前，应当先准备好它的运行环境或者源码阅读环境。若是能在实际环境下安装和运行Spark，显然可以提高读者对于Spark的一些感觉，对系统能有个大致的印象，有经验的工程师甚至可以猜出一些Spark在实现过程当中采用的设计模式、编程模型。

      考虑到大部分公司在开发和生产环境都采用Linux操做系统，因此笔者选用了64位的Linux。在正式安装Spark以前，先要找台好机器。为何？由于笔者在安装、编译、调试的过程当中发现Spark很是耗费内存，若是机器配置过低，恐怕会跑不起来。Spark的开发语言是Scala，而Scala须要运行在JVM之上，于是搭建Spark的运行环境应该包括JDK和Scala。

      本文只介绍最基本的与Spark相关的准备工做，至于Spark在实际生产环境下的配置，则须要结合具体的应用场景进行准备。

安装JDK

      自Spark2.0.0版本开始，Spark已经准备放弃对Java 7的支持，因此咱们须要选择Java 8。咱们还须要使用命令getconf LONG_BIT查看linux机器是32位仍是64位，而后下载相应版本的JDK并安装。

下载地址：

http://www.oracle.com/technetwork/java/javase/downloads/index.html。

配置环境：

cd ~
vim .bash_profile

添加以下配置：

exportJAVA_HOME=/opt/java
exportPATH=$PATH:$JAVA_HOME/bin
exportCLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

输入如下命令使环境变量快速生效：

source .bash_profile

安装完毕后，使用java –version命令查看，确认安装正常，如图1所示。

图1 查看java安装是否正常

安装Scala

      因为从Spark 2.0.0开始，Spark默认使用Scala 2.11来编译、打包，再也不是之前的Scala 2.10，因此咱们须要下载Scala 2.11。

    下载地址：

    http://www.scala-lang.org/download/

选择Scala 2.11的版本进行下载，下载方法以下：

wget https://downloads.lightbend.com/scala/2.11.8/scala-2.11.8.tgz

移动到选好的安装目录，例如：

mv scala-2.11.8.tgz~/install/

进入安装目录，执行如下命令：

chmod 755scala-2.11.8.tgz
tar -xzvfscala-2.11.8.tgz 

配置环境：

cd ~
vim .bash_profile

添加以下配置：

export SCALA_HOME=$HOME/install/scala-2.11.8
export PATH=$SCALA_HOME/bin:$PATH

输入如下命令使环境变量快速生效：

source .bash_profile

安装完毕后键入scala，进入scala命令行以确认安装正常，如图2所示。

图2 进入Scala命令行

安装Spark

      Spark进入2.0时代以后，目前一共有两个大的版本：一个是2.0.0，一个是2.1.0。本书选择2.1.0。

下载地址：

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

下载方法以下：

wget http://d3kbcqa49mib13.cloudfront.net/spark-2.1.0-bin-hadoop2.6.tgz

移动到选好的安装目录，如：

mv spark-2.1.0-bin-hadoop2.6.tgz~/install/

进入安装目录，执行如下命令：

chmod 755 spark-2.1.0-bin-hadoop2.6.tgz
tar -xzvf spark-2.1.0-bin-hadoop2.6.tgz

配置环境：

cd ~
vim .bash_profile

添加以下配置：

    export SPARK_HOME=$HOME/install/spark-2.1.0-bin-hadoop2.6
    export PATH=$SPARK_HOME/bin:$PATH

输入如下命令使环境变量快速生效：

source .bash_profile

安装完毕后键入spark-shell，进入scala命令行以确认安装正常，如图3所示。

图3 执行spark-shell进入Scala命令行

既然已经介绍了如何准备好基本的Spark运行环境，如今是时候实践一下，以便于在使用过程当中提高读者对于Spark最直接的感触！本文经过Spark的基本使用，让读者对Spark能有初步的认识，便于引导读者逐步深刻学习。

运行spark-shell

　　在《Spark2.1.0——运行环境准备》一文曾经简单运行了spark-shell，并用下图进行了展现（此处再次展现此图）。

图4    执行spark-shell进入Scala命令行

图4中显示了不少信息，这里进行一些说明：

• 在安装完Spark 2.1.0后，若是没有明确指定log4j的配置，那么Spark会使用core模块的org/apache/spark/目录下的log4j-defaults.properties做为log4j的默认配置。log4j-defaults.properties指定的Spark日志级别为WARN。用户能够到Spark安装目录的conf文件夹下从log4j.properties.template复制一份log4j.properties文件，并在其中增长本身想要的配置。
• 除了指定log4j.properties文件外，还能够在spark-shell命令行中经过sc.setLogLevel(newLevel)语句指定日志级别。
• SparkContext的Web UI的地址是：http://192.168.0.106:4040。192.168.0.106是笔者安装Spark的机器的ip地址，4040是SparkContext的Web UI的默认监听端口。
• 指定的部署模式（即master）为local[*]。当前应用（Application）的ID为local-1497084620457。
• 能够在spark-shell命令行经过sc使用SparkContext，经过spark使用SparkSession。sc和spark实际分别是SparkContext和SparkSession在Spark REPL中的变量名，具体细节已在《Spark2.1.0——剖析spark-shell》一文有过度析。

　　因为Spark core的默认日志级别是WARN，因此看到的信息不是不少。如今咱们将Spark安装目录的conf文件夹下的log4j.properties.template以以下命令复制出一份： 

cp log4j.properties.template log4j.properties

并将log4j.properties中的log4j.logger.org.apache.spark.repl.Main=WARN修改成log4j.logger.org.apache.spark.repl.Main=INFO，而后咱们再次运行spark-shell，将打印出更丰富的信息，如图5所示。

图5  Spark启动过程打印的部分信息

从图5展现的启动日志中咱们能够看到SecurityManager、SparkEnv、BlockManagerMasterEndpoint、DiskBlockManager、MemoryStore、SparkUI、Executor、NettyBlockTransferService、BlockManager、BlockManagerMaster等信息。它们是作什么的？刚刚接触Spark的读者只须要知道这些信息便可，具体内容将在后边的博文给出。

执行word count

      这一节，咱们经过word count这个耳熟能详的例子来感觉下Spark任务的执行过程。启动spark-shell后，会打开Scala命令行，而后按照如下步骤输入脚本：

步骤1    

      输入val lines =sc.textFile("../README.md", 2)，以Spark安装目录下的README.md文件的内容做为word count例子的数据源，执行结果如图6所示。

图6   步骤1执行结果

图6告诉咱们lines的实际类型是MapPartitionsRDD。

步骤2

       textFile方法对文本文件是逐行读取的，咱们须要输入val words =lines.flatMap(line => line.split(" "))，将每行文本按照空格分隔以获得每一个单词，执行结果如图7所示。

图7   步骤2执行结果

图7告诉咱们lines在通过flatMap方法的转换后获得的words的实际类型也是MapPartitionsRDD。

步骤3

     对于获得的每一个单词，经过输入val ones = words.map(w => (w,1))，将每一个单词的计数初始化为1，执行结果如图8所示。

图8   步骤3执行结果

图8告诉咱们words在通过map方法的转换后获得的ones的实际类型也是MapPartitionsRDD。

步骤4

    输入val counts = ones.reduceByKey(_ + _)，对单词进行计数值的聚合，执行结果如图9所示。

图9   步骤4执行结果

图9告诉咱们ones在通过reduceByKey方法的转换后获得的counts的实际类型是ShuffledRDD。

步骤5

       输入counts.foreach(println)，将每一个单词的计数值打印出来，做业的执行过程如图10和图11所示。做业的输出结果如图12所示。

图10   步骤5执行过程第一部分

图11  步骤5执行过程第二部分

图10和图11展现了不少做业提交、执行的信息，这里挑选关键的内容进行介绍：

• SparkContext为提交的Job生成的ID是0。
• 一共有四个RDD，被划分为ResultStage和ShuffleMapStage。ShuffleMapStage的ID为0，尝试号为0。ResultStage的ID为1，尝试号也为0。在Spark中，若是Stage没有执行完成，就会进行屡次重试。Stage不管是首次执行仍是重试都被视为是一次Stage尝试（Stage Attempt），每次Attempt都有一个惟一的尝试号（AttemptNumber）。
• 因为Job有两个分区，因此ShuffleMapStage和ResultStage都有两个Task被提交。每一个Task也会有屡次尝试，于是也有属于Task的尝试号。从图中看出ShuffleMapStage中的两个Task和ResultStage中的两个Task的尝试号也都是0。
• HadoopRDD则用于读取文件内容。

图12  步骤5输出结果

 

图12展现了单词计数的输出结果和最后打印的任务结束的日志信息。

       本文介绍的word count例子是以SparkContext的API来实现的，读者朋友们也能够选择在spark-shell中经过运用SparkSession的API来实现。

有了对Spark的初次体验，下面能够来分析下spark-shell的实现原理了，请看——《Spark2.1.0——剖析spark-shell》

想要对Spark源码进行阅读的同窗，能够看看《Spark2.1.0——代码结构及载入Ecplise方法》

关于《Spark内核设计的艺术 架构设计与实现》

通过近一年的准备，基于Spark2.1.0版本的《Spark内核设计的艺术 架构设计与实现》一书现已出版发行，图书如图：

 

纸质版售卖连接以下：

京东： https://item.jd.com/12302500.html