如何添加Hadoop依赖通过Maven

Hadoop开发中需要用到至少不下10个的依赖包，它们相互间的依赖关系比较复杂，不同版本的依赖关系也有所不同，而间接依赖导致的程序错误并不会在运行之前报错，因此确定适合一个版本的依赖包，会耗费相当多的时间。Maven是一个依赖管理和项目构建的工具，它利用惯例组织Java项目的结构，并允许通过简单的配置定义直接依赖，而直接依赖所需的其他依赖则会通过事先定义好的关系列表自动下载，大大节省了开发者本身的工作量。

使用Maven管理依赖，相比于下载到lib文件夹再导入项目，有以下优势：

节省找寻合适依赖所需的时间

便于依赖升级以适配不同的Hadoop版本

依赖下载到本地库后可重复使用

缩小版本管理的文件体积

Maven在主流的Java IDE上都有插件，Eclipse可通过安装m2eclipse，但这里个人推荐与智能化程度更高、联网更迅速、体验更流畅的IntelliJ IDEA配合，以下就简要介绍IntelliJ IDEA如何使用Maven。（以IntelliJ IDEA 13为例，它自带了Maven的支持，无需安装插件）

New Project ->Maven新建一个Maven项目，在属性栏填写GroupId（项目或组织的唯一标识）和ArtifactId（项目的通用名称）：

Next之后填写项目名和项目位置：

新建项目之后有d窗提示，选择自动导入：

项目生成后的架构是这样的，我们在src/main/java路径下写项目代码，pom.xml是项目统一的配置文件。

因为我们线上用的是Cloudera的Hadoop版本，所以我们需要在pom.xml的project节点下添加一个CDH5的Maven远程依赖仓库（参考Using the CDH 5 Maven Repository)

开发一个普通的Hadoop项目，我们一般需要hadoop-common、hadoop-core两组依赖；如果需要读取HDFS上的文件内容，则需要hadoop-hdfs和hadoop-client另外两组依赖；如果需要读取HBase的数据，则需要再加入hbase-client。（以上均为artifactId，以CDH5.1.0为例）

IntelliJ IDEA提供了对pom文件非常智能的自动补全和实时查询功能：

创建 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 的模式运行。

---