==Spark快速入门

Spark快速入门 | Ji ZHANG&＃8217;s Blog
http://shzhangji.com/blog/2014/12/16/spark-quick-start/

sc.textFile()用于生成一个RDD，并声明该RDD指向的是/tmp/logs.txt文件。RDD可以暂时认为是一个列表，列表中的元素是一行行日志（因此是String类型）。

lines.map(f)表示对RDD中的每一个元素使用f函数来处理，并返回一个新的RDD。
line => line.split(“\t”)是一个匿名函数，又称为Lambda表达式、闭包等。它的作用和普通的函数是一样的，如这个匿名函数的参数是line（String类型），返回值是Array数组类型，因为String.split()函数返回的是数组。

需要注意的是，cache函数并不会立刻执行缓存操作，事实上map、filter等函数都不会立刻执行，而是在用户执行了一些特定操作后才会触发，比如first、count、reduce等。这两类操作分别称为Transformations和Actions。

之后对firstTenErrors的处理使用的是Scala集合类库中的方法，如map、foreach，和RDD提供的接口基本一致。所以说用Scala编写Spark程序是最自然的。

Apache Spark是新兴的一种快速通用的大规模数据处理引擎。它的优势有三个方面：
通用计算引擎 能够运行MapReduce、数据挖掘、图运算、流式计算、SQL等多种框架；
基于内存 数据可缓存在内存中，特别适用于需要迭代多次运算的场景；
与Hadoop集成 能够直接读写HDFS中的数据，并能运行在YARN之上。

Spark是用Scala语言编写的，所提供的API也很好地利用了这门语言的特性。它也可以使用Java和Python编写应用。本文将用Scala进行讲解。
安装Spark和SBT
从官网上下载编译好的压缩包，解压到一个文件夹中。下载时需注意对应的Hadoop版本，如要读写CDH4 HDFS中的数据，则应下载Pre-built for CDH4这个版本。
为了方便起见，可以将spark/bin添加到$PATH环境变量中：

1
2

export SPARK_HOME=/path/to/spark
export PATH=$PATH:$SPARK_HOME/bin

在练习例子时，我们还会用到SBT这个工具，它是用来编译打包Scala项目的。Linux下的安装过程比较简单：下载sbt-launch.jar到$HOME/bin目录；
新建$HOME/bin/sbt文件，权限设置为755，内容如下：

1
2

SBT_OPTS=&＃8221;-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=256M&＃8221;
java $SBT_OPTS -jar dirname $0/sbt-launch.jar &＃8220;$@&＃8221;

日志分析示例
假设我们有如下格式的日志文件，保存在/tmp/logs.txt文件中：
1
2
3
4
5

2014-12-11 18:33:52 INFO Java some message
2014-12-11 18:34:33 INFO MySQL some message
2014-12-11 18:34:54 WARN Java some message
2014-12-11 18:35:25 WARN Nginx some message
2014-12-11 18:36:09 INFO Java some message

每条记录有四个字段，即时间、级别、应用、信息，使用制表符分隔。
Spark提供了一个交互式的命令行工具，可以直接执行Spark查询：
1
2
3
4
5
6
7
8
9

$ spark-shell
Welcome to

/ / ___ _____/ /__
\ / _ / _ `/ / &＃8216;/
// ./_,// //_\ version 1.1.0
/_/
Spark context available as sc.
scala>

加载并预览数据
1
2
3
4
5

scala> val lines = sc.textFile(&＃8220;/tmp/logs.txt&＃8221;)
lines: org.apache.spark.rdd.RDD[String] = /tmp/logs.txt MappedRDD[1] at textFile at :12

scala> lines.first()
res0: String = 2014-12-11 18:33:52 INFO Java some message

sc是一个SparkContext类型的变量，可以认为是Spark的入口，这个对象在spark-shell中已经自动创建了。
sc.textFile()用于生成一个RDD，并声明该RDD指向的是/tmp/logs.txt文件。RDD可以暂时认为是一个列表，列表中的元素是一行行日志（因此是String类型）。这里的路径也可以是HDFS上的文件，如hdfs://127.0.0.1:8020/user/hadoop/logs.txt。
lines.first()表示调用RDD提供的一个方法：first()，返回第一行数据。

解析日志
为了能对日志进行筛选，如只处理级别为ERROR的日志，我们需要将每行日志按制表符进行分割：
1
2
3
4
5

scala> val logs = lines.map(line => line.split(&＃8220;\t&＃8221;))
logs: org.apache.spark.rdd.RDD[Array[String]] = MappedRDD[2] at map at :14

scala> logs.first()
res1: Array[String] = Array(2014-12-11 18:33:52, INFO, Java, some message)

lines.map(f)表示对RDD中的每一个元素使用f函数来处理，并返回一个新的RDD。
line => line.split(“\t”)是一个匿名函数，又称为Lambda表达式、闭包等。它的作用和普通的函数是一样的，如这个匿名函数的参数是line（String类型），返回值是Array数组类型，因为String.split()函数返回的是数组。
同样使用first()方法来看这个RDD的首条记录，可以发现日志已经被拆分成四个元素了。

过滤并计数
我们想要统计错误日志的数量：
1
2
3
4
5
6
7
8

scala> val errors = logs.filter(log => log(1) == &＃8220;ERROR&＃8221;)
errors: org.apache.spark.rdd.RDD[Array[String]] = FilteredRDD[3] at filter at :16

scala> errors.first()
res2: Array[String] = Array(2014-12-11 18:39:42, ERROR, Java, some message)

scala> errors.count()
res3: LOng= 158

logs.filter(f)表示筛选出满足函数f的记录，其中函数f需要返回一个布尔值。
log(1) == “ERROR”表示获取每行日志的第二个元素（即日志级别），并判断是否等于ERROR。
errors.count()用于返回该RDD中的记录。

缓存
由于我们还会对错误日志做一些处理，为了加快速度，可以将错误日志缓存到内存中，从而省去解析和过滤的过程：
1

scala> errors.cache()

errors.cache()函数会告知Spark计算完成后将结果保存在内存中。所以说Spark是否缓存结果是需要用户手动触发的。在实际应用中，我们需要迭代处理的往往只是一部分数据，因此很适合放到内存里。
需要注意的是，cache函数并不会立刻执行缓存操作，事实上map、filter等函数都不会立刻执行，而是在用户执行了一些特定操作后才会触发，比如first、count、reduce等。这两类操作分别称为Transformations和Actions。
显示前10条记录
1
2
3
4
5
6
7

scala> val firstTenErrors = errors.take(10)
firstTenErrors: Array[Array[String]] = Array(Array(2014-12-11 18:39:42, ERROR, Java, some message), Array(2014-12-11 18:40:23, ERROR, Nginx, some message), &＃8230;)

scala> firstTenErrors.map(log => log.mkString(&＃8220;\t&＃8221;)).foreach(line => println(line))
2014-12-11 18:39:42 ERROR Java some message
2014-12-11 18:40:23 ERROR Nginx some message
&＃8230;

errors.take(n)方法可用于返回RDD前N条记录，它的返回值是一个数组。之后对firstTenErrors的处理使用的是Scala集合类库中的方法，如map、foreach，和RDD提供的接口基本一致。所以说用Scala编写Spark程序是最自然的。
按应用进行统计
我们想要知道错误日志中有几条Java、几条Nginx，这和常见的Wordcount思路是一样的。
1
2
3
4
5
6
7
8
9
10
11
12
13

scala> val apps = errors.map(log => (log(2), 1))
apps: org.apache.spark.rdd.RDD[(String, Int)] = MappedRDD[15] at map at :18

scala> apps.first()
res20: (String, Int) = (Java,1)

scala> val counts = apps.reduceByKey((a, b) => a + b)
counts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[17] at reduceByKey at :20

scala> counts.foreach(t => println(t))
(Java,58)
(Nginx,53)
(MySQL,47)

errors.map(log => (log(2), 1))用于将每条日志转换为键值对，键是应用（Java、Nginx等），值是1，如(&＃8220;Java&＃8221;, 1)
，这种数据结构在Scala中称为元组（Tuple），这里它有两个元素，因此称为二元组。
对于数据类型是二元组的RDD，Spark提供了额外的方法，reduceByKey(f)就是其中之一。它的作用是按键进行分组，然后对同一个键下的所有值使用f函数进行归约（reduce）。归约的过程是：使用列表中第一、第二个元素进行计算，然后用结果和第三元素进行计算，直至列表耗尽。如：
1
2

scala> Array(1, 2, 3, 4).reduce((a, b) => a + b)
res23: Int = 10

上述代码的计算过程即((1 + 2) + 3) + 4
。
counts.foreach(f)表示遍历RDD中的每条记录，并应用f函数。这里的f函数是一条打印语句（println）。
打包应用程序
为了让我们的日志分析程序能够在集群上运行，我们需要创建一个Scala项目。项目的大致结构是：
1
2
3
4
5
6
7
8
9

spark-sandbox
├── build.sbt
├── project
│ ├── build.properties
│ └── plugins.sbt
└── src
└── main
└── scala
└── LogMining.scala

你可以直接使用这个项目作为模板。下面说明一些关键部分：
配置依赖
build.sbt

1

libraryDependencies += &＃8220;org.apache.spark&＃8221; %% &＃8220;spark-core&＃8221; % &＃8220;1.1.1&＃8221;

程序内容
src/main/scala/LogMining.scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

object LogMining extends App {
val cOnf= new SparkConf().setAppName(&＃8220;LogMining&＃8221;)
val sc = new SparkContext(conf)
val inputFile = args(0)
val lines = sc.textFile(inputFile)
// 解析日志
val logs = lines.map(.split(&＃8220;\t&＃8221;))
val errors = logs.filter((1) == &＃8220;ERROR&＃8221;)
// 缓存错误日志
errors.cache()
// 统计错误日志记录数
println(errors.count())
// 获取前10条MySQL的错误日志
val mysqlErrors = errors.filter((2) == &＃8220;MySQL&＃8221;)
mysqlErrors.take(10).map( mkString &＃8220;\t&＃8221;).foreach(println)
// 统计每个应用的错误日志数
val errorApps = errors.map(_(2) -> 1)
errorApps.countByKey().foreach(println)
}

打包运行
1
2
3

$ cd spark-sandbox
$ sbt package
$ spark-submit &＃8211;class LogMining &＃8211;master local target/scala-2.10/spark-sandbox_2.10-0.1.0.jar data/logs.txt

参考资料
Spark Programming Guide
Introduction to Spark Developer Training
Spark Runtime Internals