天天看点

CentOS7下安装配置Spark-2.0.1集群

Spark是一个快速、通用的计算集群框架,它的内核使用Scala语言编写,它提供了Scala、Java和Python编程语言high-level API,使用这些API能够非常容易地开发并行处理的应用程序。

下面,我们通过搭建Spark集群计算环境,并进行简单地验证,来体验一下使用Spark计算的特点。无论从安装运行环境还是从编写处理程序(用Scala,Spark默认提供的Shell环境可以直接输入Scala代码进行数据处理),我们都会觉得比Hadoop MapReduce计算框架要简单得多,而且,Spark可以很好地与HDFS进行交互(从HDFS读取数据,以及写数据到HDFS中)。

安装配置

下载安装配置Scala

<code>wget http:</code><code>//www</code><code>.scala-lang.org</code><code>/files/archive/scala-2</code><code>.10.3.tgz</code>

<code>tar</code> <code>xvzf scala-2.10.3.tgz</code>

在~/.bashrc中增加环境变量SCALA_HOME,并使之生效:

<code>export</code> <code>SCALA_HOME=</code><code>/usr/scala/scala-2</code><code>.10.3</code>

<code>export</code> <code>PATH=$PATH:$SCALA_HOME</code><code>/bin</code>

下载安装配置Spark

我们首先在主节点m1上配置Spark程序,然后将配置好的程序文件复制分发到集群的各个从结点上。下载解压缩:

<code>wget http:</code><code>//d3kbcqa49mib13</code><code>.cloudfront.net</code><code>/spark-0</code><code>.9.0-incubating-bin-hadoop1.tgz</code>

<code>tar</code> <code>xvzf spark-0.9.0-incubating-bin-hadoop1.tgz</code>

在~/.bashrc中增加环境变量SPARK_HOME,并使之生效:

<code>export</code> <code>SPARK_HOME=</code><code>/home/shirdrn/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1</code>

<code>export</code> <code>PATH=$PATH:$SPARK_HOME</code><code>/bin</code>

在m1上配置Spark,修改spark-env.sh配置文件:

<code>cd</code> <code>/home/shirdrn/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1</code><code>/conf</code>

<code>cp</code> <code>spark-</code><code>env</code><code>.sh.template spark-</code><code>env</code><code>.sh</code>

在该脚本文件中,同时将SCALA_HOME配置为Unix环境下实际指向路径,例如:

修改conf/slaves文件,将计算节点的主机名添加到该文件,一行一个,例如:

<code>s1</code>

<code>s2</code>

<code>s3</code>

最后,将Spark的程序文件和配置文件拷贝分发到从节点机器上:

<code>scp</code> <code>-r ~</code><code>/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1 shirdrn@s1:~</code><code>/cloud/programs/</code>

<code>scp</code> <code>-r ~</code><code>/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1 shirdrn@s2:~</code><code>/cloud/programs/</code>

<code>scp</code> <code>-r ~</code><code>/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1 shirdrn@s3:~</code><code>/cloud/programs/</code>

启动Spark集群

我们会使用HDFS集群上存储的数据作为计算的输入,所以首先要把Hadoop集群安装配置好,并成功启动,我这里使用的是Hadoop 1.2.1版本。启动Spark计算集群非常简单,执行如下命令即可:

<code>cd</code> <code>/home/shirdrn/cloud/programs/spark-0</code><code>.9.0-incubating-bin-hadoop1/</code>

<code>sbin</code><code>/start-all</code><code>.sh</code>

可以看到,在m1上启动了一个名称为Master的进程,在s1上启动了一个名称为Worker的进程,如下所示,我这里也启动了Hadoop集群:

主节点m1上:

<code>54968 SecondaryNameNode</code>

<code>55651 Master</code>

<code>54814 NameNode</code>

从节点s1上:

<code>33592 Worker</code>

<code>33442 TaskTracker</code>

<code>33336 DataNode</code>

各个进程是否启动成功,也可以查看日志来诊断,例如:

主节点上:

<code>tail</code> <code>-100f $SPARK_HOME</code><code>/logs/spark-shirdrn-org</code><code>.apache.spark.deploy.master.Master-1-m1.out</code>

从节点上:

<code>tail</code> <code>-100f $SPARK_HOME</code><code>/logs/spark-shirdrn-org</code><code>.apache.spark.deploy.worker.Worker-1-s1.out</code>

Spark集群计算验证

我们使用我的网站的访问日志文件来演示,示例如下:

<code>27.159.254.192 - - [21</code><code>/Feb/2014</code><code>:11:40:46 +0800] </code><code>"GET /archives/526.html HTTP/1.1"</code> <code>200 12080 </code><code>"http://shiyanjun.cn/archives/526.html"</code> <code>"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"</code>

<code>120.43.4.206 - - [21</code><code>/Feb/2014</code><code>:10:37:37 +0800] </code><code>"GET /archives/417.html HTTP/1.1"</code> <code>200 11464 </code><code>"http://shiyanjun.cn/archives/417.html/"</code> <code>"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"</code>

统计该文件里面IP地址出现频率,来验证Spark集群能够正常计算。另外,我们需要从HDFS中读取这个日志文件,然后统计IP地址频率,最后将结果再保存到HDFS中的指定目录。

首先,需要启动用来提交计算任务的Spark Shell:

<code>bin</code><code>/spark-shell</code>

在Spark Shell上只能使用Scala语言写代码来运行。

然后,执行统计IP地址频率,在Spark Shell中执行如下代码来实现:

<code>val </code><code>file</code> <code>= sc.textFile(</code><code>"hdfs://m1:9000/user/shirdrn/wwwlog20140222.log"</code><code>)</code>

<code>val result = </code><code>file</code><code>.flatMap(line =&gt; line.</code><code>split</code><code>(</code><code>"\\s+.*"</code><code>)).map(word =&gt; (word, 1)).reduceByKey((a, b) =&gt; a + b)</code>

上述的文件hdfs://m1:9000/user/shirdrn/wwwlog20140222.log是输入日志文件。处理过程的日志信息,示例如下所示:

<code> </code> 

<code>14</code><code>/03/06</code> <code>21:59:23 INFO Executor: Running task ID 20</code>

<code>36</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO BlockManager: Found block broadcast_11 locally</code>

<code>37</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Getting 1 non-zero-bytes blocks out of 1 blocks</code>

<code>38</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 0 remote gets </code><code>in</code>  <code>1 ms</code>

<code>39</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO Executor: Serialized size of result </code><code>for</code> <code>20 is 19423</code>

<code>40</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO Executor: Sending result </code><code>for</code> <code>20 directly to driver</code>

<code>41</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO TaskSetManager: Finished TID 20 </code><code>in</code> <code>17 ms on localhost (progress: 0</code><code>/1</code><code>)</code>

<code>42</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO TaskSchedulerImpl: Remove TaskSet 20.0 from pool</code>

<code>43</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO DAGScheduler: Completed ResultTask(20, 0)</code>

<code>44</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO DAGScheduler: Stage 20 (collect at &lt;console&gt;:13) finished </code><code>in</code> <code>0.016 s</code>

<code>45</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO SparkContext: Job finished: collect at &lt;console&gt;:13, took 0.242136929 s</code>

<code>46</code>

<code>14</code><code>/03/06</code> <code>21:59:23 INFO Executor: Finished task ID 20</code>

<code>47</code>

<code>res14: Array[(String, Int)] = Array((27.159.254.192,28), (120.43.9.81,40), (120.43.4.206,16), (120.37.242.176,56), (64.31.25.60,2), (27.153.161.9,32), (202.43.145.163,24), (61.187.102.6,1), (117.26.195.116,12), (27.153.186.194,64), (123.125.71.91,1), (110.85.106.105,64), (110.86.184.182,36), (27.150.247.36,52), (110.86.166.52,60), (175.98.162.2,20), (61.136.166.16,1), (46.105.105.217,1), (27.150.223.49,52), (112.5.252.6,20), (121.205.242.4,76), (183.61.174.211,3), (27.153.230.35,36), (112.111.172.96,40), (112.5.234.157,3), (144.76.95.232,7), (31.204.154.144,28), (123.125.71.22,1), (80.82.64.118,3), (27.153.248.188,160), (112.5.252.187,40), (221.219.105.71,4), (74.82.169.79,19), (117.26.253.195,32), (120.33.244.205,152), (110.86.165.8,84), (117.26.86.172,136), (27.153.233.101,8), (123.12...</code>

可以看到,输出了经过map和reduce计算后的部分结果。

最后,我们想要将结果保存到HDFS中,只要输入如下代码:

result.saveAsTextFile("hdfs://m1:9000/user/shirdrn/wwwlog20140222.log.result")

查看HDFS上的结果数据:

<code>[shirdrn@m1 ~]$ hadoop fs -</code><code>cat</code> <code>/user/shirdrn/wwwlog20140222</code><code>.log.result</code><code>/part-00000</code> <code>| </code><code>head</code> <code>-5</code>

<code>(27.159.254.192,28)</code>

<code>(120.43.9.81,40)</code>

<code>(120.43.4.206,16)</code>

<code>(120.37.242.176,56)</code>

<code>(64.31.25.60,2)</code>

<code></code>

      本文转自crazy_charles 51CTO博客,原文链接:http://blog.51cto.com/douya/1863288,如需转载请自行联系原作者