Spark本地模式部署与实例运行

Spark 本地模式

实验环境

• Spark 1.6.1
• Hadoop 2.6.4
• JDK7
• Scala2.10.2
• Ubuntu14.04-64bit
  Hadoop伪分布式安装具体过程参照Hadoop伪分布式安装。
  Scala2.11版本需要重新编译Spark。

安装Spark

1. 解压
   此处解压到/home/下
2. 修改配置文件
   安装后，需要在 ./conf/spark-env.sh 中修改 Spark 的 Classpath，执行如下命令拷贝一个配置文件：

cp ./conf/spark-env.sh.template ./conf/spark-env.sh

编辑./conf/spark-env.sh（vim ./conf/spark-env.sh)，在最后面加上如下内容：

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)

1. 检查安装是否成功
   进入spark目录下的bin/文件夹，执行spark-shell命令

4.运行Wordcount实例
启动hdfs，在hdfs中新建test/目录，将spark目录下的README.md文件复制到test/下。在spark-shell中执行如下 scala语句：

scala> val input = sc.textFile("hdfs://master:9000/user/spark/test/README.md"）
scala> val words = input.flatMap(x => x.split(" "))
scala> val reslut = words.map(x => (x,1)).reduceByKey((x,y) => x+y)

以上语句的解释可以参考官方文档：

• 从程序中数据创建RDD
• 从外部数据，如HDFS中创建RDD
• RDD基本操作

执行结果如下：

5.运行pagerank实例

1. Intellij IDEA中新建scala工程，在project
   structure中为src/目录下新建main/scala目录，设置为sources类型，并添加scala和spark-assembly-xxx-hadoopxxx.jar两个Library
2. 新建test包，添加pageranktest对象文件，代码见PageRank code：
3. 打包程序
   在项目结构界面中选择"Artifacts"，在右边操作界面选择绿色"+"号，选择添加JAR包的"From modules with dependencies"方式，出现如下界面，在该界面中选择主函数入口；
   点击菜单Build->Build Artifacts，弹出选择动作，选择Build或者Rebuild动作；
   复制打包文件到Spark根目录下；
4. 通过spark-submit运行程序
   本地模式下提交程序的例子：

## Run application locally on 8 cores
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master local[8] \
  /path/to/examples.jar \
  100

将打包的PageRank程序提交

spark-submit --master local[2] --class tset.pageranktest ~/spark-1.6.1/pagerank.jar 

按照Rank值排序打印

PageRank code

package tset
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
object pageranktest {
    def main(args: Array[String]) {
      //屏蔽日志
      Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
      Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
      //设置运行环境
      val conf = new SparkConf().setAppName("PageRank").setMaster("local")
      val sc = new SparkContext(conf)
      //读入数据文件
      val articles: RDD[String] = sc.textFile("hdfs://master:9000/user/spark/test/graphx-wiki-vertices.txt")
      val links: RDD[String] = sc.textFile("hdfs://master:9000/user/spark/test/graphx-wiki-edges.txt")
      //装载顶点和边
      val vertices = articles.map { line =>
        val fields = line.split('\t')
        (fields(0).toLong, fields(1))
      }
      val edges = links.map { line =>
        val fields = line.split('\t')
        Edge(fields(0).toLong, fields(1).toLong, 0)
      }
      val graph = Graph(vertices, edges, "").persist() 
      println("**********************************************************")
      println("PageRank计算，获取最有价值的数据")
      println("**********************************************************")
      val prGraph = graph.pageRank(0.001).cache()
      val titleAndPrGraph = graph.outerJoinVertices(prGraph.vertices) {
        (v, title, rank) => (rank.getOrElse(0.0), title)
      }
      titleAndPrGraph.vertices.top(10) {
        Ordering.by((entry: (VertexId, (Double, String))) => entry._2._1)
      }.foreach(t => println(t._2._2 + ": " + t._2._1))
      sc.stop()
    }
  }