Spark学习笔记8-在eclipse里用Java编写spark程序

0.前言

这次的学习内容是，在eclipse里用java编写spark程序。相关jar包用maven来管理。程序实现笔记7一样的内容，做一个README.md文件的单词计数，并打印出来，这次只做单机下运行，集群模式可以参考Spark学习笔记7的内容完成。

1.安装eclipse

1.1下载eclipse

下载地址如下：
http://www.eclipse.org/downloads/
我的是64位的mac，所以下载j2e，64位的。

1.2运行

直接运行就行。

2.创建maven工程

2.1创建工程

点击file－>new->Maven Project

直接next

选择maven-archetype-quickstart

创建Group ID，Artifact ID，直接finish完成。

成功后是这样的：（这里要注意了，如果创建不能成功，报错Could not resolve archetype org.apache.maven.archetypes:maven-archetype-quickstart，解决方法看博客：http://blog.csdn.net/xummgg/article/details/50655382 ）

上面的JRE System Libary里面默认的不是1.8版本，这个要自己进入build path里面的configure build path修改，修改成想要的版本。

2.2 修改pom.xml文件

在创建好的工程里有个pom.xml文件，修改该文件。在maven中pom.xml文件是用来配置所有jar包的。

修改好后内容如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>com.dt.spark</groupId>
<artifactId>SparkApps</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>SparkApps</name>
<url>http://maven.apache.org</url>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>  
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-graphx_2.10</artifactId>
<version>1.6.0</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/java</sourceDirectory>
<testSourceDirectory>src/main/test</testSourceDirectory>

<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass></mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>

<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<executable>java</executable>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<classpathScope>compile</classpathScope>
<mainClass>com.dt.spark.App</mainClass>
</configuration>
</plugin>

<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>

</plugins>
</build>
</project>

如何修改这个pom文件可以看下面2个网站：
可以在spark官网上找到各个组件的maven配置方法：
http://spark.apache.org/docs/latest/

也可以在maven的下面网址中找到关于spark的配置方法：
http://maven.outofmemory.cn/org.apache.spark/

2.3 保存pom.xml文件

按Alt＋S保存该文件，可以看到如下图，右下脚开始有进度条，自动下载tar包。

完成后可以看到所有jar包被导入：

3.编写程序并运行

终于可以开始用java来编写spark程序啦。
创建新工程SparkApps.cores，再创建新类WordCount.java

然后编写程序（程序的具体意思，可以参考Spark学习笔记7的代码注释，内容是一样的，只是用不同语言实现）：

package com.dt.spark.SparkApps.cores;

import java.util.Arrays;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/**
 * 使用Java的方式开发进行本地测试Spark的WordCount程序
 */
public class WordCount {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("Spakr WordCount written by jaba").setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> lines = sc.textFile("/Users/xxm/Documents/soft/spark-1.6.0-bin-hadoop2.6/README.md");

        JavaRDD<String> words =lines.flatMap(new FlatMapFunction<String, String>() {

            public Iterable<String> call(String line) throws Exception {
                // TODO Auto-generated method stub

                return Arrays.asList(line.split(" "));
            }
        });

        JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

            public Tuple2<String, Integer> call(String word) throws Exception {
                // TODO Auto-generated method stub

                return new Tuple2<String,Integer>(word,1);
            }
        });

        JavaPairRDD<String, Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

            public Integer call(Integer v1, Integer v2) throws Exception {
                // TODO Auto-generated method stub
                return v1 + v2;
            }
        });

        wordsCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {

            public void call(Tuple2<String, Integer> pairs) throws Exception {
                // TODO Auto-generated method stub
                System.out.println(pairs._1+":"+pairs._2);
            }
        });

        sc.close();
    }

}

写好后，运行java程序，结果如下。Wordcount完成啦。

---

XianMing