使用spark-submit运行Spark程序 2015-03-13 15:00

安装sbt

sbt的安装非常简单,下载,解压,配置$PATH环境变量即可。

准备目录

sbt的目录结构如下:

$ find .
.
./hello.sbt
./src
./src/main
./src/main/scala
./src/main/scala/hello.scala

hello.sbt的内容为:

注意:里面的空行不能删除

name := "Hello Project"

version := "1.0"

scalaVersion := "2.11.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.2.1"

hello.scala的内容为:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

object HelloApp {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Hello Application")
    val sc = new SparkContext(conf)
    /* 读入HDFS文件 */
    val logRDD = sc.textFile("hdfs://ctrl:9000/spark/test/1.log")
    /* 从每行第一列提取出IP地址字符串, 并将每个记录形成 (IP,1)的形式 */
    val ipRDD = logRDD.map( s => (s.split(" ",2)(0), 1))
    /* 以每一个IP为key,进行规约求总数 */
    val countRDD = ipRDD.reduceByKey((a, b) => a + b)
    /* 按每个(IP,n)元组的第2列进行倒序排序 */
    val sortRDD = countRDD.sortBy( c => c._2, false)
    /* 将结果输出到HDFS文件中 */
    sortRDD.saveAsTextFile("hdfs://ctrl:9000/spark/test/output3")

    println(sortRDD.collect())
  }
}

打包

此步骤需要联网下载相关软件,时间较长。

sbt package

运行

spark-submit \
  --class "HelloApp" \
  --master yarn-cluster \
  target/scala-2.11/hello-project_2.11-1.0.jar
Tags: #Spark    Post on Spark