如何在CDH5上运行Spark应用

创建 maven 工程

使用下面命令创建一个普通的 maven 工程：

bash

$ mvn archetype:generate -DgroupId=com.cloudera.sparkwordcount -DartifactId=sparkwordcount -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

将 sparkwordcount 目录重命名为simplesparkapp，然后，在 simplesparkapp 目录下添加 scala 源文件目录：

bash

$ mkdir -p sparkwordcount/src/main/scala/com/cloudera/sparkwordcount

修改 pom.xml 添加 scala 和 spark 依赖：

xml

<dependencies>

<dependency>

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

<version>2.10.4</version>

</dependency>

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.10</artifactId>

<version>1.2.0-cdh5.3.0</version>

</dependency>

</dependencies>

添加编译 scala 的插件：

xml

<plugin>

<groupId>org.scala-tools</groupId>

<artifactId>maven-scala-plugin</artifactId>

<executions>

<execution>

<goals>

<goal>compile</goal>

<goal>testCompile</goal>

</goals>

</execution>

</executions>

</plugin>

添加 scala 编译插件需要的仓库：

xml

<pluginRepositories>

<pluginRepository>

<id>scala-tools.org</id>

<name>Scala-tools Maven2 Repository</name>

<url>http://scala-tools.org/repo-releases</url>

</pluginRepository>

</pluginRepositories>

另外，添加 cdh hadoop 的仓库：

xml

<repositories>

<repository>

<id>scala-tools.org</id>

<name>Scala-tools Maven2 Repository</name>

<url>http://scala-tools.org/repo-releases</url>

</repository>

<repository>

<id>maven-hadoop</id>

<name>Hadoop Releases</name>

<url>https://repository.cloudera.com/content/repositories/releases/</url>

</repository>

<repository>

<id>cloudera-repos</id>

<name>Cloudera Repos</name>

<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>

</repository>

</repositories>

最后，完整的 pom.xml 文件见： https://github.com/javachen/simplesparkapp/blob/master/pom.xml 。

运行下面命令检查工程是否能够成功编译：

bash

mvn package

编写示例代码

以 WordCount 为例，该程序需要完成以下逻辑：

读一个输入文件

统计每个单词出现次数

过滤少于一定次数的单词

对剩下的单词统计每个字母出现次数

在 MapReduce 中，上面的逻辑需要两个 MapReduce 任务，而在 Spark 中，只需要一个简单的任务，并且代码量会少 90%。

编写 Scala 程序 如下：

scala

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

object SparkWordCount {

def main(args: Array[String]) {

val sc = new SparkContext(new SparkConf().setAppName("Spark Count"))

val threshold = args(1).toInt

// split each document into words

val tokenized = sc.textFile(args(0)).flatMap(_.split(" "))

// count the occurrence of each word

val wordCounts = tokenized.map((_, 1)).reduceByKey(_ + _)

// filter out words with less than threshold occurrences

val filtered = wordCounts.filter(_._2 >= threshold)

// count characters

val charCounts = filtered.flatMap(_._1.toCharArray).map((_, 1)).reduceByKey(_ + _)

System.out.println(charCounts.collect().mkString(", "))

charCounts.saveAsTextFile("world-count-result")

}

}

Spark 使用懒执行的策略，意味着只有当 动作 执行的时候， 转换 才会运行。上面例子中的 动作 *** 作是 collect 和 saveAsTextFile ，前者是将数据推送给客户端，后者是将数据保存到 HDFS。

作为对比， Java 版的程序 如下：

java

import java.util.ArrayList

import java.util.Arrays

import org.apache.spark.api.java.*

import org.apache.spark.api.java.function.*

import org.apache.spark.SparkConf

import scala.Tuple2

public class JavaWordCount {

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("Spark Count"))

final int threshold = Integer.parseInt(args[1])

// split each document into words

JavaRDD tokenized = sc.textFile(args[0]).flatMap(

new FlatMapFunction() {

public Iterable call(String s) {

return Arrays.asList(s.split(" "))

}

}

)

// count the occurrence of each word

JavaPairRDD counts = tokenized.mapToPair(

new PairFunction() {

public Tuple2 call(String s) {

return new Tuple2(s, 1)

}

}

).reduceByKey(

new Function2() {

public Integer call(Integer i1, Integer i2) {

return i1 + i2

}

}

)

另外， Python 版的程序 如下：

python

import sys

from pyspark import SparkContext

file="inputfile.txt"

count=2

if __name__ == "__main__":

sc = SparkContext(appName="PythonWordCount")

lines = sc.textFile(file, 1)

counts = lines.flatMap(lambda x: x.split(' ')) \

.map(lambda x: (x, 1)) \

.reduceByKey(lambda a, b: a + b) \

.filter(lambda (a, b) : b >= count) \

.flatMap(lambda (a, b): list(a)) \

.map(lambda x: (x, 1)) \

.reduceByKey(lambda a, b: a + b)

print ",".join(str(t) for t in counts.collect())

sc.stop()

编译

运行下面命令生成 jar：

bash

$ mvn package

运行成功之后，会在 target 目录生成 sparkwordcount-0.0.1-SNAPSHOT.jar 文件。

运行

因为项目依赖的 spark 版本是 1.2.0-cdh5.3.0 ，所以下面的命令只能在 CDH 5.3 集群上运行。

首先，将测试文件 inputfile.txt 上传到 HDFS 上；

bash

$ wget https://github.com/javachen/simplesparkapp/blob/master/data/inputfile.txt

$ hadoop fs -put inputfile.txt

其次，将 sparkwordcount-0.0.1-SNAPSHOT.jar 上传到集群中的一个节点；然后，使用 spark-submit 脚本运行 Scala 版的程序：

bash

$ spark-submit --class com.cloudera.sparkwordcount.SparkWordCount --master local sparkwordcount-0.0.1-SNAPSHOT.jar inputfile.txt 2

或者，运行 Java 版本的程序：

bash

$ spark-submit --class com.cloudera.sparkwordcount.JavaWordCount --master local sparkwordcount-0.0.1-SNAPSHOT.jar inputfile.txt 2

对于 Python 版的程序，运行脚本为：

bash

$ spark-submit --master local PythonWordCount.py

如果，你的集群部署的是 standalone 模式，则你可以替换 master 参数的值为 spark://<master host>:<master port>，也可以以 Yarn 的模式运行。

sqoop是map任务，分布到集群所有机器上执行。

所以你的授权（grant）只给了''@'localhost'是不够的。你至少需要将权限给到'user'@'%'，这其中的user，是你访问mysql使用的用户。

---