大数据Spark：动手写WordCount

问题导读

1.Spark集群的执行单位是什么？
2.RDD的创建的来源是什么？
3.RDD之间是否可以转换？
4.如何实现修改spark WordCount？






Spark是主流的大数据处理框架，具体有啥能耐，相信不需要多说。我们开门见山，直接动手写大数据界的HelloWorld：WordCount。

先上完整代码，看看咋样能入门。

[mw_shl_code=python,true]import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object WordCount {
  def main(args: Array[String]) {
    val conf = new SparkConf()
    conf.setAppName("WordCount")
    val sc = new SparkContext(conf)

    val file = "hdfs://127.0.0.1:9000/file.txt"
    val lines = sc.textFile(file)
    val words = lines.flatMap(_.split("\\s+"))
    val wordCount = words.countByValue()

    println(wordCount)
  }
}[/mw_shl_code]


寥寥10多行代码，就已经完成了，比大家想象的要简单，完全看不出大数据背后的存储，分布式，容错处理，这就是Spark给我们带来的福利。

接下来我们逐步解析其中的核心概念。

Spark上下文

Spark集群的执行单位是Application，任何提交的任务都会产生一个Application。一个Application只会关联上一个Spark上下文，也就是SparkContext。构建SparkContext时可以传入Spark相关配置，也就是SparkConf，它可以用来指定Application的名称，任务需要的CPU核数/内存大小，调优需要的配置等等。

[mw_shl_code=python,true]val conf = new SparkConf()
conf.setAppName("WordCount")
val sc = new SparkContext(conf)[/mw_shl_code]


这三行语句创建了一个Spark上下文，并且运行时这个Application的名字就叫WordCount。

弹性分布式数据集RDD

Spark中最主要的编程概念就是弹性分布式数据集 (resilient distributed dataset,RDD)，它是元素的集合，划分到集群的各个节点上，可以被并行操作。RDD的创建可以从HDFS(或者任意其他支持Hadoop文件系统) 上的一个文件开始，或者通过转换Master中已存在的Scala集合而来。

[mw_shl_code=python,true]val file = "hdfs://127.0.0.1:9000/file.txt"
val lines = sc.textFile(file)[/mw_shl_code]

这两行语句从hdfs文件中创建了叫lines的RDD，它的每个元素就对应文件中的每一行，有了RDD我们就可以通过它提供的各种API来完成需要的业务功能。

RDD提供的API分为两类：转换（Transformation）和动作（Action）。

转换

顾名思义，转换就是把一个RDD转换成另一个RDD。当然，光是拷贝产生一个新的RDD是没有太大意义的，这里的转换实际上是RDD中元素的映射和转换。有一点必须要注意的是，RDD是只读的，一旦执行转换，一定会生成一个新的RDD。

[mw_shl_code=python,true]val words = lines.flatMap(_.split("\\s+"))[/mw_shl_code]

flatMap是RDD众多转换中的一种，它的功能是把源RDD中的元素映射成目的RDD中的0个或者多个元素。上面语句把以文本行为元素的RDD转换成了以单个单词为元素的RDD。

动作

“动作”就不好望文生义了，可以简单地理解成想要获得结果时调用的API。

[mw_shl_code=python,true]val wordCount = words.countByValue()[/mw_shl_code]

countByValue就是一个“动作”，它的功能是统计RDD中每个元素出现的次数，最终得到一个元素及其出现次数的Map。

那究竟哪些API是转换，哪些是动作呢？有个很简单的判断准则：

提示：返回结果为RDD的API是转换，返回结果不为RDD的API是动作。

运行

要运行Spark任务，首先要把代码打成JAR包，额。。。这个不需要多言。

打包后，就只需在Spark集群上以命令行的方式用spark-submit提交就OK。

[mw_shl_code=bash,true]spark-submit --class "demo.WordCount" SparkDemo-1.0-SNAPSHOT.jar[/mw_shl_code]

其中demo.WordCount是main函数所在的ojbect，而SparkDemo-1.0-SNAPSHOT.jar就是打出来的jar包。

大数据处理，就这样入门了。


优雅程序员 原创

来源：http://www.cnblogs.com/yoyaprogrammer/p/spark_wordcount.html

不错，简洁

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

这两个语句在我的Intellij IDEA中总是出现 Unused import statement ,  spark的assembly , scala库已增加了，不知什么原因，麻烦高人指点。