開(kāi)發(fā)wordcount程序
1朦佩、用Java開(kāi)發(fā)wordcount程序
1.1 配置maven環(huán)境
1.2 如何進(jìn)行本地測(cè)試
1.3 如何使用spark-submit提交到spark集群進(jìn)行執(zhí)行（spark-submit常用參數(shù)說(shuō)明冯遂，spark-submit其實(shí)就類(lèi)似于hadoop的hadoop jar命令）拟枚。spark-submit封裝在shell腳本里面
2颂郎、用Scala開(kāi)發(fā)wordcount程序
2.1 下載scala ide for eclipse
2.2 在Java Build Path中吼渡，添加spark依賴包（如果與scala ide for eclipse原生的scala版本發(fā)生沖突，則移除原生的scala / 重新配置scala compiler）
2.3 用export導(dǎo)出scala spark工程
3乓序、用spark-shell開(kāi)發(fā)wordcount程序
3.1 常用于簡(jiǎn)單的測(cè)試
<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>cn.spark</groupId>
<artifactId>SparkTest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>SparkTest</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.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.10</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.4.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.3.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>cn.spark.sparktest.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>
WordCountLocal.java
package cn.spark.study.core;
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開(kāi)發(fā)本地測(cè)試的wordcount程序
*
*/
public class WordCountLocal {
public static void main(String[] args) {
// 編寫(xiě)Spark應(yīng)用程序
// 本地執(zhí)行寺酪，是可以執(zhí)行在eclipse中的main方法中，執(zhí)行的
// 第一步：創(chuàng)建SparkConf對(duì)象替劈，設(shè)置Spark應(yīng)用的配置信息
// 使用setMaster()可以設(shè)置Spark應(yīng)用程序要連接的Spark集群的master節(jié)點(diǎn)的url
// 但是如果設(shè)置為local則代表寄雀，在本地運(yùn)行
SparkConf conf = new SparkConf()
.setAppName("WordCountLocal")
.setMaster("local");
// 第二步：創(chuàng)建JavaSparkContext對(duì)象
// 在Spark中，SparkContext是Spark所有功能的一個(gè)入口陨献，你無(wú)論是用java盒犹、scala，甚至是python編寫(xiě)
// 都必須要有一個(gè)SparkContext眨业，它的主要作用阿趁，包括初始化Spark應(yīng)用程序所需的一些核心組件，包括
// 調(diào)度器（DAGSchedule坛猪、TaskScheduler）脖阵，還會(huì)去到Spark Master節(jié)點(diǎn)上進(jìn)行注冊(cè)，等等
// 一句話墅茉，SparkContext命黔，是Spark應(yīng)用中呜呐，可以說(shuō)是最最重要的一個(gè)對(duì)象
// 但是呢，在Spark中悍募，編寫(xiě)不同類(lèi)型的Spark應(yīng)用程序蘑辑，使用的SparkContext是不同的，如果使用scala坠宴，
// 使用的就是原生的SparkContext對(duì)象
// 但是如果使用Java洋魂，那么就是JavaSparkContext對(duì)象
// 如果是開(kāi)發(fā)Spark SQL程序，那么就是SQLContext喜鼓、HiveContext
// 如果是開(kāi)發(fā)Spark Streaming程序副砍，那么就是它獨(dú)有的SparkContext
// 以此類(lèi)推
JavaSparkContext sc = new JavaSparkContext(conf);
// 第三步：要針對(duì)輸入源（hdfs文件、本地文件庄岖，等等）豁翎，創(chuàng)建一個(gè)初始的RDD
// 輸入源中的數(shù)據(jù)會(huì)打散，分配到RDD的每個(gè)partition中隅忿，從而形成一個(gè)初始的分布式的數(shù)據(jù)集
// 我們這里呢心剥，因?yàn)槭潜镜販y(cè)試，所以呢背桐，就是針對(duì)本地文件
// SparkContext中优烧，用于根據(jù)文件類(lèi)型的輸入源創(chuàng)建RDD的方法，叫做textFile()方法
// 在Java中链峭，創(chuàng)建的普通RDD匙隔，都叫做JavaRDD
// 在這里呢，RDD中熏版，有元素這種概念纷责，如果是hdfs或者本地文件呢，創(chuàng)建的RDD撼短，每一個(gè)元素就相當(dāng)于
// 是文件里的一行
JavaRDD<String> lines = sc.textFile("C://Users//Administrator//Desktop//spark.txt");
// 第四步：對(duì)初始RDD進(jìn)行transformation操作再膳，也就是一些計(jì)算操作
// 通常操作會(huì)通過(guò)創(chuàng)建function，并配合RDD的map曲横、flatMap等算子來(lái)執(zhí)行
// function喂柒，通常，如果比較簡(jiǎn)單禾嫉，則創(chuàng)建指定Function的匿名內(nèi)部類(lèi)
// 但是如果function比較復(fù)雜灾杰，則會(huì)單獨(dú)創(chuàng)建一個(gè)類(lèi)，作為實(shí)現(xiàn)這個(gè)function接口的類(lèi)
// 先將每一行拆分成單個(gè)的單詞
// FlatMapFunction熙参，有兩個(gè)泛型參數(shù)艳吠，分別代表了輸入和輸出類(lèi)型
// 我們這里呢，輸入肯定是String孽椰，因?yàn)槭且恍幸恍械奈谋菊衙洌敵隽莞荩鋵?shí)也是String，因?yàn)槭敲恳恍械奈谋? // 這里先簡(jiǎn)要介紹flatMap算子的作用栏渺，其實(shí)就是呛梆，將RDD的一個(gè)元素，給拆分成一個(gè)或多個(gè)元素
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});
// 接著磕诊，需要將每一個(gè)單詞填物，映射為(單詞, 1)的這種格式
// 因?yàn)橹挥羞@樣，后面才能根據(jù)單詞作為key霎终，來(lái)進(jìn)行每個(gè)單詞的出現(xiàn)次數(shù)的累加
// mapToPair滞磺，其實(shí)就是將每個(gè)元素，映射為一個(gè)(v1,v2)這樣的Tuple2類(lèi)型的元素
// 如果大家還記得scala里面講的tuple神僵，那么沒(méi)錯(cuò)，這里的tuple2就是scala類(lèi)型覆劈，包含了兩個(gè)值
// mapToPair這個(gè)算子保礼，要求的是與PairFunction配合使用，第一個(gè)泛型參數(shù)代表了輸入類(lèi)型
// 第二個(gè)和第三個(gè)泛型參數(shù)责语，代表的輸出的Tuple2的第一個(gè)值和第二個(gè)值的類(lèi)型
// JavaPairRDD的兩個(gè)泛型參數(shù)炮障，分別代表了tuple元素的第一個(gè)值和第二個(gè)值的類(lèi)型
JavaPairRDD<String, Integer> pairs = words.mapToPair(
new PairFunction<String, String, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, Integer> call(String word) throws Exception {
return new Tuple2<String, Integer>(word, 1);
}
});
// 接著，需要以單詞作為key坤候，統(tǒng)計(jì)每個(gè)單詞出現(xiàn)的次數(shù)
// 這里要使用reduceByKey這個(gè)算子胁赢，對(duì)每個(gè)key對(duì)應(yīng)的value，都進(jìn)行reduce操作
// 比如JavaPairRDD中有幾個(gè)元素白筹，分別為(hello, 1) (hello, 1) (hello, 1) (world, 1)
// reduce操作智末，相當(dāng)于是把第一個(gè)值和第二個(gè)值進(jìn)行計(jì)算，然后再將結(jié)果與第三個(gè)值進(jìn)行計(jì)算
// 比如這里的hello徒河，那么就相當(dāng)于是系馆，首先是1 + 1 = 2，然后再將2 + 1 = 3
// 最后返回的JavaPairRDD中的元素顽照，也是tuple由蘑，但是第一個(gè)值就是每個(gè)key，第二個(gè)值就是key的value
// reduce之后的結(jié)果代兵，相當(dāng)于就是每個(gè)單詞出現(xiàn)的次數(shù)
JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(
new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
// 到這里為止尼酿，我們通過(guò)幾個(gè)Spark算子操作，已經(jīng)統(tǒng)計(jì)出了單詞的次數(shù)
// 但是植影，之前我們使用的flatMap裳擎、mapToPair、reduceByKey這種操作思币，都叫做transformation操作
// 一個(gè)Spark應(yīng)用中句惯，光是有transformation操作土辩，是不行的，是不會(huì)執(zhí)行的抢野，必須要有一種叫做action
// 接著拷淘，最后，可以使用一種叫做action操作的指孤，比如說(shuō)启涯，foreach，來(lái)觸發(fā)程序的執(zhí)行
wordCounts.foreach(new VoidFunction<Tuple2<String,Integer>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<String, Integer> wordCount) throws Exception {
System.out.println(wordCount._1 + " appeared " + wordCount._2 + " times.");
}
});
sc.close();
}
}
WordCount.scala
package cn.spark.study.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
/**
* @author Administrator
*/
object WordCount {
def main(args: Array[String]) {
val conf = new SparkConf()
.setAppName("WordCount");
val sc = new SparkContext(conf)
val lines = sc.textFile("hdfs://spark1:9000/spark.txt", 1);
val words = lines.flatMap { line => line.split(" ") }
val pairs = words.map { word => (word, 1) }
val wordCounts = pairs.reduceByKey { _ + _ }
wordCounts.foreach(wordCount => println(wordCount._1 + " appeared " + wordCount._2 + " times."))
}
}
