一、pom
<properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.version>2.12.10</scala.version> <spark.version>3.0.0</spark.version> <hadoop.version>3.2.1</hadoop.version> <encoding>UTF-8</encoding> </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.2.1</version> </dependency> </dependencies> <build> <pLuginManagement> <plugins> <!--编译scala的插件-- -> <pLugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> </plugin> <!--编译java的插件--> <pLugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> </plugin> </plugins> </pLuginManagement> <plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <executions> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>add-source</goal> <goal>compile</goal> </goals> </execution> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.1.1</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass></mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build>
二、Spark3.0-JavaAPI程序
实现Spark读取HDFS中的文本文件，实现单词计数，并2将结果输出到HDFS中。
2.1 java匿名实现类
// 1.创建配置 SparkConf conf = SparkConf().setAppName("JavaWordCount"); // 2.封装了SparkContext JavaSparkContext jsc = new JavaSparkContext(conf); // 3.jsc 创建 java RDD JavaRDD<String> lines = jsc.textFile(args[0]); // 4.切分压平 JavaRDD<String> words = lines.flatMap(new FlatMapFuntion<String,String>(){ @Override public Iterator<String> call(String line) throws Expection{ return Arrays.asList(line.split(" ")).iterator; } }); // 5.单词和1组合 JavaPairRDD<String,Integer> wordAndOne = words.maoToPair(new PairFunction<String,String,Integer>(){ @Override public Tuple2<String,Integer> call(String words) throws Exception{ return Tuple2.apply(word,1); } }); // 6.分组聚合 JavaPairRDD<String,Integer> reduced = wordAndOne.reduceByKey(new Function2<Integer,Integer,Integer>(){ @Override public Integer call(Integer v1,Integer v2) throws Exception{ return v1+v2; } }); // 7.交换kv顺序 JavaPairRDD<Integer,String> swapped = reduced.mapToPair(new PairFunction<Tuple2<String,Integer>,Integer,String>(){ @Override public Tuple2<Integer,String> call(Tuple2<String,Integer> tp) throws Exception{ return tp.swap(); } }); // 8.排序 JavaPairRDD<Integer,String> sorted = swapped.sortedByKey(false); // 9.交换kv顺序 JavaPairRDD<String,Integer> result = sorted.mapToPair(new PairFunction<Tuple2<String,Integer>,String,Integer>(){ @Override public Tuple2<String,Integer> call(Tuple2<Integer,String> tp) throws Exception{ return tp.swap(); } }); // 10.触发action保存到HDFS result.saveAsTextFile(args[1]); // 11.释放资源 jsc.stop();
2.2 Lambda表达式实现
// 1.创建配置 SparkConf conf = SparkConf().setAppName("LambdaJavaWordCount"); // 2.封装了SparkContext JavaSparkContext jsc = new JavaSparkContext(conf); // 3.jsc 创建 java RDD JavaRDD<String> lines = jsc.textFile(args[0]); // 4.切分压平 JavaRDD<String> words = lines.flatMap(line -> Arrays.stream(line.split(" ")).iterator()); // 5.单词 1 JavaPairRDD<String,Integer> wordAndOne = words.mapToPair(w -> Tuple2.apply(w,1)); // 6.聚合 JavaPairRDD<String,Integer> reduced = wordAndOne.reduceByKey((i,j) -> i+j); // 7.排序 JavaPairRDD<String,Integer> sorted = reduced.mapToPair(tp -> tp.swap()).sortByKey(false) .mapToPair(tp -> tp.swap()); // 8.保存hdfs sorted.saveAsTextFile(args[1]); // 9.释放资源 jsc.stop();
2.3 程序打包
2.4 上传到Linux
2.5 启动HDFS
hadoop-daemon.sh start namenode
hadoop-daemon.sh start datanode
2.6 Spark执行jar包
bin/spark-3.0.0-bin-hadoop3.2/bin/spark-submit --master spark://hadoop1:7077,hadoop2:7077,hadoop3:7077 --executor-memory 1g --total-executor-cores 5 --class com.wang.spark.LambdaJavaWordCount /root/spark-in-active-1.0-SNAPSHOT.jar hdfs://hadoop1:9000/wc hdfs://hadoop1:9000/out
2.7 查看结果
hdfs -dfs -cat /out/*
三、本机执行
本地测试，不会建立集群链接，再本地的一个进程运行。
// 1.创建配置【本地测试】 SparkConf conf = SparkConf().setAppName("LambdaJavaWordCount").setMaster("local[*]"); // 2.封装了SparkContext JavaSparkContext jsc = new JavaSparkContext(conf); // 3.jsc 创建 java RDD JavaRDD<String> lines = jsc.textFile(args[0]); // 4.切分压平 JavaRDD<String> words = lines.flatMap(line -> Arrays.stream(line.split(" ")).iterator()); // 5.单词 1 JavaPairRDD<String,Integer> wordAndOne = words.mapToPair(w -> Tuple2.apply(w,1)); // 6.聚合 JavaPairRDD<String,Integer> reduced = wordAndOne.reduceByKey((i,j) -> i+j); // 7.排序 JavaPairRDD<String,Integer> sorted = reduced.mapToPair(tp -> tp.swap()).sortByKey(false) .mapToPair(tp -> tp.swap()); // 8.保存hdfs sorted.saveAsTextFile(args[1]); // 9.释放资源 jsc.stop();
运行时候，传入参数本地数据或者hdfs的数据。
如果出现这个错误，需要将pom中的scala的****放开
或者全部读取本机的文件。
执行结果如下：
