0. 前言
搭好的Spark当然要先写一个最简单的WordCount练练手。
那么，需求是：
1、统计Spark下README.md文件的词频；
2、输出较多，筛选出现次数超过10次的，词频逆序显示
注意：
- 一般用hdfs上的文件，这里为了简化，用spark目录中的文件。
- 读取hdfs用hdfs协议，读取文件用file协议，不写协议默认用file协议读取当前目录下文件。
- 要分布式集群上用file读取文件的话，Worker节点的该路径下都需要有这个文件。
- 尝鲜而已，怎么简单怎么来，先读取本地文件好了：
4.1 以本地模式运行，设置setMaster("local")；
4.2 local可以接中括号[]，括号中的数字表示执行任务的线程数；
4.3 local[*] 表示CPU有几个核就用几个线程。
0.1 构建工程：
这里可以直接构建scala工程，但是一次准备写scala和java的，因此从基础的maven项目开始构建。
0.1.1 Maven quickstart
0.0.1 GroupId+ArtifactId+项目初始化确认点点点
略
0.0.2 添加Scala Framework 支持
-
右键点击项目名，选择Add Framework，选择Scala
image.png -
我已经写过scala项目了，这里默认库选好了。初始化可能不是这个版本，需要创建，点击Use library最右边的Create：
image.png 确认下拉框中的版本，和Compiler classpath，Standard library中依赖包的版本一致，如果不一致，需要先去pom文件中添加对应版本的依赖，把包虾虾来之后点击这里下面的+号进行依赖的选择
百度有说去Project Structure 的Platform Setting 的Global Library里添加，直接下载。放弃吧，你下不下来的，你下不下来的，你下不下来的。我一开始只在pom写了spark的核心依赖，看着maven下包的时候发现一闪而过的Hadoop的默认依赖下的2.6的，又去pom中手动指定了一下。
1. Scala
构建scala工程
package com.junzerg
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
def main(args: Array[String]): Unit = {
/**
* 获取编程入口
*/
val conf: SparkConf = new SparkConf()
conf.setAppName("WordCount")
conf.setMaster("local")
val sc: SparkContext = new SparkContext(conf)
/**
* 通过编程入口加载数据
*/
val linesRdd:RDD[String] = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md")
/**
* 对数据进行处理
*/
val wordCountRdd = linesRdd
.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)
.filter(_._2 > 10)
.filter(_._1 != "")
.sortBy(_._2, false)
/**
* 对结果数据进行处理l
*/
wordCountRdd.foreach(println(_))
sc.stop()
}
}
1.1 遇到的问题
首次运行报错，百度又是一通鬼扯：
Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 10582
打断点，看异常栈，发现最后抛出的异常在：
declaringClass = "com.thoughtworks.paranamer.BytecodeReadingParanamer$ClassReader"
methodName = "accept"
fileName = "BytecodeReadingParanamer.java"
lineNumber = 563
这个类默认下载了2.7版本。和jdk8兼容不是很好，要去pom指定一下：
<dependency>
<groupId>com.thoughtworks.paranamer</groupId>
<artifactId>paranamer</artifactId>
<version>2.8</version>
</dependency>
2. Java7（无lambda表达式java）
在IDEA和pom中，退一下版本
/**
* @author pengjunzhe
*/
public class WordCountJdk7 {
private static final Pattern SPACE = Pattern.compile(" ");
public static void main(String[] args) {
/**
* 获取编程入口
*/
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName("WordCountJdk7");
JavaSparkContext sc = new JavaSparkContext(conf);
/**
* 通过编程入口加载数据
*/
JavaRDD<String> lines = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md", 2);
/**
* 对数据进行处理
*/
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterator<String> call(String s) {
return Arrays.asList(s.split(" ")).iterator();
}
});
JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<>(s, 1);
}
});
ones = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
}).filter(new Function<Tuple2<String, Integer>, Boolean>() {
@Override
public Boolean call(Tuple2<String, Integer> v1) throws Exception {
return v1._2 > 10;
}
}).filter(new Function<Tuple2<String, Integer>, Boolean>() {
@Override
public Boolean call(Tuple2<String, Integer> v1) throws Exception {
return v1._1.length() > 1;
}
});
JavaPairRDD<Integer, String> transdOnes = ones.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
@Override
public Tuple2<Integer, String> call(Tuple2<String, Integer> t2) throws Exception {
return new Tuple2<>(t2._2, t2._1);
}
}).sortByKey();
ones = transdOnes.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
@Override
public Tuple2<String, Integer> call(Tuple2<Integer, String> t2) throws Exception {
return new Tuple2<>(t2._2, t2._1);
}
});
/**
* 对结果数据进行处理
*/
ones.foreach(new VoidFunction<Tuple2<String, Integer>>() {
@Override
public void call(Tuple2<String, Integer> t2) throws Exception {
System.out.println(t2);
}
});
/**
* 结束关闭入口
*/
sc.stop();
}
}
3. Java8（带lambda表达式java）
/**
* @author pengjunzhe
*/
public class WordCountJdk8 {
public static void main(String[] args) {
/**
* 获取编程入口
*/
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("WordCountJdk8");
JavaSparkContext sc = new JavaSparkContext(conf);
/**
* 通过编程入口加载数据
*/
JavaRDD<String> lines = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md", 2);
/**
* 对数据进行处理
*/
JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
JavaPairRDD<String, Integer> ones = words
.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1))
.reduceByKey((v1, v2) -> v1 + v2)
.filter(v1 -> v1._2 > 10)
.filter(v1 -> v1._1.length() > 1);
JavaPairRDD<Integer, String> transdOnes = ones
.mapToPair(t2 -> new Tuple2<>(t2._2, t2._1))
.sortByKey();
ones = transdOnes.mapToPair(t2 -> new Tuple2<>(t2._2, t2._1));
/**
* 对结果数据进行处理
*/
// ones.foreach(System.out::println);
ones.foreach(s -> System.out.println(s));
/**
* 结束关闭入口
*/
sc.stop();
}
}
注意最后不能用System.out::println的写法，抛出PrintStream不能被序列化的异常。
4.Python
from pyspark import SparkConf, SparkContext
# 获取编程入口
conf = SparkConf() \
.setMaster("local[*]") \
.setAppName("WordCount")
sc = SparkContext(conf=conf)
# 通过编程入口加载数据
linesRdd = sc.textFile("file:///Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md")
# 对数据进行处理
wordCountRdd = linesRdd \
.flatMap(lambda x: x.split(" ")) \
.map(lambda x: (x, 1)) \
.reduceByKey(lambda x, y: x + y) \
.filter(lambda x: x[1] > 10) \
.filter(lambda x: x[0] != "" ) \
.sortBy(lambda x: x[1], False)
# 对结果数据进行处理
wordCountRdd.foreach(lambda x: print(x))
# 关闭编程
sc.stop()
