使用Spark完成WordCount程序（Java、Scala双语言）

小鱼儿 2022-09-05 04:19 197阅读 0赞

我们分别使用scala和java两门语言完成项目

先创建一个maven项目，添加java和scala的sdk

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x0MzI2MDMwNDM0_size_16_color_FFFFFF_t_70]

#### Scala版本 ####

pom文件安装依赖

<dependency>
    	<groupId>org.apache.spark</groupId>
    	<artifactId>spark-code_2.11</artifactId>
    	<version>2.4.3</version>
    </dependency>

import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * 需求:单词计数 */
    object WordCountScala { 
    
    	def main(args: Array[String]): Unit = {  
    	// 第一步:创建SparkContext
    	val conf = new SparkConf() 
    	conf.setAppName("WordCountScala") //设置任务名称
    		.setMaster("local") //local表示在本地执行 
    	val sc = new SparkContext(conf)
    	// 第二步:加载数据
    	val linesRDD = sc.textFile("~/hello.txt")
    	// 第三步:对数据进行切割，把一行数据切分成一个一个的单词 
    	val wordsRDD = linesRDD.flatMap(_.split(" "))
    	// 第四步:迭代words,将每个word转化为(word,1)这种形式 
    	val pairRDD = wordsRDD.map((_,1))
    	// 第五步:根据key(其实就是word)进行分组聚合统计 
    	val wordCountRDD = pairRDD.reduceByKey(_ + _)
    	// 第六步:将结果打印到控制台 
    	wordCountRDD.foreach(wordCount=>println(wordCount._1+"--"+wordCount._2))
    	// 第七步:停止SparkContext
    	sc.stop() 
    	}
    }

由于此时我们在代码中设置的Master为local，表示会在本地创建一个临时的spark集群运 行这个代码，这样有利于代码调试

执行代码，结果如下:

> you–1  
> hello–2  
> me–1

我们再来总结一下代码中这几个RDD中的数据结构

val linesRDD = sc.textFile("~/hello.txt")

linesRDD中的数据是这样的:

> hello you  
> hello me

val wordsRDD = linesRDD.flatMap(_.split(" "))

wordsRDD中的数据是这样的:

> hello  
> you  
> hello  
> me

val pairRDD = wordsRDD.map((_,1))

pairRDD 中的数据是这样的 (hello,1)

> (you,1)  
> (hello,1)  
> (me,1)

val wordCountRDD = pairRDD.reduceByKey(_ + _)

wordCountRDD 中的数据是这样的

> (hello,2)  
> (you,1)  
> (me,1)

这是Scala代码的实现，接下来我们用java代码实现一下

#### Java版本 ####

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.*;
    import scala.Tuple2;
    import java.util.Arrays;
    import java.util.Iterator;
    /** * 需求:单词计数 */
    public class WordCountJava { 
        public static void main(String[] args) { 
    		// 第一步:创建SparkContext: 
    		// 注意，针对java代码需要获取JavaSparkContext 
    		SparkConf conf = new SparkConf(); 
    		conf.setAppName("WordCountJava").setMaster("local");
    		JavaSparkContext sc = new JavaSparkContext(conf);
    		
    		// 第二步:加载数据
    		JavaRDD<String> linesRDD = sc.textFile("~/hello.txt");
    		// 第三步:对数据进行切割，把一行数据切分成一个一个的单词 
    		// 注意:FlatMapFunction的泛型，第一个参数表示输入数据类型，第二个表示是输 
    		JavaRDD<String> wordRDD = linesRDD.flatMap(new FlatMapFunction<String, String>() { 
    	        public Iterator<String> call(String line) throws Exception { 
    	            return Arrays.asList(line.split(" ")).iterator();
    	        } 
    	    });
    	    
    		// 第四步:迭代words,将每个word转化为(word,1)这种形式
    		// 注意:PairFunction的泛型，第一个参数是输入数据类型
    		// 第二个是输出tuple中的第一个参数类型，第三个是输出tuple中的第二个参数类型
    		// 注意:如果后面需要使用到....ByKey，前面都需要使用mapToPair去处理
    		JavaPairRDD<String, Integer> pairRDD = wordRDD.mapToPair(new PairFunction<Strinig, String, Integer>(){ 
    			public Tuple2<String, Integer> call(String word) throws Exception { 
    				return new Tuple2<String, Integer>(word, 1);
    			}             
    	    })
    		// 第五步:根据key(其实就是word)进行分组聚合统计
    		JavaPairRDD<String, Integer> wordCountRDD = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() { 
    	        public Integer call(Integer i1, Integer i2) throws Exception { 
    	            return i1 + i2;
    			} 
    		});
    		
    		// 第六步:将结果打印到控制台
    		wordCountRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() { 
    	        public void call(Tuple2<String, Integer> tup) throws Exception { 
    	            System.out.println(tup._1+"--"+tup._2);
    			} 
    		});
    		
    		// 第七步:停止sparkContext
    		sc.stop();
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x0MzI2MDMwNDM0_size_16_color_FFFFFF_t_70]: /images/20220829/1a9e4da5de1c442e930c2d7b2a6844b5.png