spark从入门到放弃一:worldcountjava

文章地址&＃xff1a;http://www.haha174.top/article/details/253584
项目源码&＃xff1a;https://github.com/haha174/spark.git
开启spark 从入门到放弃/笑哭。下面不多说来写一个hello world 压压惊。
之前搭建集群用的是 spark 2.2 hadoop 2.9
所以开发的需要引入如下的依赖

<properties><project.build.sourceEncoding>UTF-8project.build.sourceEncoding><spark.version>2.2.1spark.version><java.version>1.8java.version><hadoop.version>2.9.0hadoop.version>properties><dependencies><dependency><groupId>junitgroupId><artifactId>junitartifactId><version>4.12version><scope>testscope>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-core_2.10artifactId><version>${spark.version}version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-sql_2.10artifactId><version>${spark.version}version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-hive_2.10artifactId><version>${spark.version}version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-streaming_2.10artifactId><version>${spark.version}version>dependency><dependency><groupId>org.apache.hadoopgroupId><artifactId>hadoop-clientartifactId><version>${hadoop.version}version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-streaming-kafka_2.11artifactId><version>1.6.2version>dependency>dependencies>


/*** 第一步 &＃xff0c;创建SparkConf* setMaster 设置 集群 master的url 如果设置为local 表示在本地运行*/SparkConf conf&＃61;new SparkConf().setAppName("WorldCountLocal").setMaster("local");

/**
* 第二步 创建SparkContext 对象
* 在spark 中SparkContext 是spark 所有功能的入口 无论使用的是java scala 甚至py 编写都必须有一个SparkContext
* 它的主要作用包括初始化spark 应用程序所需要的一些核心组件&＃xff0c;包括调度器&＃xff08;DAGSchedule,taskScheduler&＃xff09;,他还会去spark master 节点上去注册等等
* 但是呢&＃xff0c;在spark中编写不同类型的spark 应用程序&＃xff0c;使用的SparkContext
* 如果使用scala 使用 使用原生的SparkContext
* 如果使用java 那么就是用JavaSparkContext
* 如果使用Spark Sql 程序 那么就是 SQLContext,HiveContext
* 如果开发Spark Streaming 程序 那么就是它独有的SparkContext
* 以此类推
*/
JavaSparkContext sc&＃61;new JavaSparkContext(conf);

/**
* 第三步&＃xff1a; 要针对输入源&＃xff08;hdfs文件,本地文件,等等&＃xff09;创建一个初始的RDD
* 输入源中的数据会被打散&＃xff0c;分配到RDD的每个partition 中从而形成一个初始的分布式数据集
* 本次测试 所以针对本地文件
* SparkContext 中用于根据文件 类型的输入源创建RDD 的方法&＃xff0c;叫做textFile()
* 在我们这里呢 RDD 中 有元素这种概念 如果是 hdfs 或者本地文件呢 创建RDD 每一个文件就相当于文件里面的一行
*/
JavaRDD lines&＃61;sc.textFile("C:\\Users\\haha174\\Desktop\\data\\world-count.txt");

// 第四步&＃xff1a;对初始RDD进行transformation操作&＃xff0c;也就是一些计算操作
// 通常操作会通过创建function&＃xff0c;并配合RDD的map、flatMap等算子来执行
// function&＃xff0c;通常&＃xff0c;如果比较简单&＃xff0c;则创建指定Function的匿名内部类
// 但是如果function比较复杂&＃xff0c;则会单独创建一个类&＃xff0c;作为实现这个function接口的类
// 先将每一行拆分成单个的单词
// FlatMapFunction&＃xff0c;有两个泛型参数&＃xff0c;分别代表了输入和输出类型
// 我们这里呢&＃xff0c;输入肯定是String&＃xff0c;因为是一行一行的文本&＃xff0c;输出&＃xff0c;其实也是String&＃xff0c;因为是每一行的文本
// 这里先简要介绍flatMap算子的作用&＃xff0c;其实就是&＃xff0c;将RDD的一个元素&＃xff0c;给拆分成一个或多个元素
JavaRDD<String> words &＃61; lines.flatMap(new FlatMapFunction<String, String>() {
public Iterator<String> call(String s) throws Exception {
return Arrays.asList(s.split("-")) .iterator();
}
});

// 接着&＃xff0c;需要将每一个单词&＃xff0c;映射为(单词, 1)的这种格式
// 因为只有这样&＃xff0c;后面才能根据单词作为key&＃xff0c;来进行每个单词的出现次数的累加
// mapToPair&＃xff0c;其实就是将每个元素&＃xff0c;映射为一个(v1,v2)这样的Tuple2类型的元素
// 如果大家还记得scala里面讲的tuple&＃xff0c;那么没错&＃xff0c;这里的tuple2就是scala类型&＃xff0c;包含了两个值
// mapToPair这个算子&＃xff0c;要求的是与PairFunction配合使用&＃xff0c;第一个泛型参数代表了输入类型
// 第二个和第三个泛型参数&＃xff0c;代表的输出的Tuple2的第一个值和第二个值的类型
// JavaPairRDD的两个泛型参数&＃xff0c;分别代表了tuple元素的第一个值和第二个值的类型
JavaPairRDD pairs &＃61; words.mapToPair(new PairFunction() {private static final long serialVersionUID &＃61; 1L;public Tuple2 call(String word) throws Exception {return new Tuple2(word, 1);}});
// 接着&＃xff0c;需要以单词作为key&＃xff0c;统计每个单词出现的次数
// 这里要使用reduceByKey这个算子&＃xff0c;对每个key对应的value&＃xff0c;都进行reduce操作
// 比如JavaPairRDD中有几个元素&＃xff0c;分别为(hello, 1) (hello, 1) (hello, 1) (world, 1)
// reduce操作&＃xff0c;相当于是把第一个值和第二个值进行计算&＃xff0c;然后再将结果与第三个值进行计算
// 比如这里的hello&＃xff0c;那么就相当于是&＃xff0c;首先是1 &＃43; 1 &＃61; 2&＃xff0c;然后再将2 &＃43; 1 &＃61; 3
// 最后返回的JavaPairRDD中的元素&＃xff0c;也是tuple&＃xff0c;但是第一个值就是每个key&＃xff0c;第二个值就是key的value
// reduce之后的结果&＃xff0c;相当于就是每个单词出现的次数
/*** public interface Function2 extends Serializable {*R call(T1 var1, T2 var2) throws Exception;*}* 第三个参数表示返回类型*/
JavaPairRDD wordCounts &＃61; pairs.reduceByKey(new Function2() {private static final long serialVersionUID &＃61; 1L;public Integer call(Integer v1, Integer v2) throws Exception {return v1 &＃43; v2;}});
// 到这里为止&＃xff0c;我们通过几个Spark算子操作&＃xff0c;已经统计出了单词的次数

// 到这里为止&＃xff0c;我们通过几个Spark算子操作&＃xff0c;已经统计出了单词的次数// 但是&＃xff0c;之前我们使用的flatMap、mapToPair、reduceByKey这种操作&＃xff0c;都叫做transformation操作// 一个Spark应用中&＃xff0c;光是有transformation操作&＃xff0c;是不行的&＃xff0c;是不会执行的&＃xff0c;必须要有一种叫做action// 接着&＃xff0c;最后&＃xff0c;可以使用一种叫做action操作的&＃xff0c;比如说&＃xff0c;foreach&＃xff0c;来触发程序的执行wordCounts.foreach(new VoidFunction>() {private static final long serialVersionUID &＃61; 1L;public void call(Tuple2 wordCount) throws Exception {System.out.println(wordCount._1 &＃43; " appeared " &＃43; wordCount._2 &＃43; " times.");}});sc.close();
}