Spark编程基础

Scala编写Spark的WorkCount

创建一个Maven项目

在pom.xml中添加依赖和插件

1. <properties>
2.     <maven.compiler.source>8</maven.compiler.source>
3.     <maven.compiler.target>8</maven.compiler.target>
4.     <encoding>UTF-8</encoding>
5.     <spark.version>3.2.3</spark.version>
6.     <scala.version>2.12.15</scala.version>
7. </properties>
9. <dependencies>
10.    
11.     <dependency>
12.         <groupId>org.scala-lang</groupId>
13.         <artifactId>scala-library</artifactId>
14.         <version>${scala.version}</version>
15.     </dependency>
17.    
18.     <dependency>
19.         <groupId>org.apache.spark</groupId>
20.         <artifactId>spark-core_2.12</artifactId>
21.         <version>${spark.version}</version>
22.     </dependency>
24. </dependencies>
28. <repositories>
29.     <repository>
30.         <id>nexus-aliyun</id>
31.         <name>Nexus aliyun</name>
32.         <layout>default</layout>
33.         <url>http://maven.aliyun.com/nexus/content/groups/public</url>
34.         <snapshots>
35.             <enabled>false</enabled>
36.             <updatePolicy>never</updatePolicy>
37.         </snapshots>
38.         <releases>
39.             <enabled>true</enabled>
40.             <updatePolicy>never</updatePolicy>
41.         </releases>
42.     </repository>
43. </repositories>
46. <pluginRepositories>
47.     <pluginRepository>
48.         <id>ali-plugin</id>
49.         <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
50.         <snapshots>
51.             <enabled>false</enabled>
52.             <updatePolicy>never</updatePolicy>
53.         </snapshots>
54.         <releases>
55.             <enabled>true</enabled>
56.             <updatePolicy>never</updatePolicy>
57.         </releases>
58.     </pluginRepository>
59. </pluginRepositories>
61. <build>
62.     <pluginManagement>
63.         <plugins>
64.             
65.             <plugin>
66.                 <groupId>net.alchim31.maven</groupId>
67.                 <artifactId>scala-maven-plugin</artifactId>
68.                 <version>3.2.2</version>
69.             </plugin>
70.             
71.             <plugin>
72.                 <groupId>org.apache.maven.plugins</groupId>
73.                 <artifactId>maven-compiler-plugin</artifactId>
74.                 <version>3.5.1</version>
75.             </plugin>
76.         </plugins>
77.     </pluginManagement>
78.     <plugins>
79.         <plugin>
80.             <groupId>net.alchim31.maven</groupId>
81.             <artifactId>scala-maven-plugin</artifactId>
82.             <executions>
83.                 <execution>
84.                     <id>scala-compile-first</id>
85.                     <phase>process-resources</phase>
86.                     <goals>
87.                         <goal>add-source</goal>
88.                         <goal>compile</goal>
89.                     </goals>
90.                 </execution>
91.                 <execution>
92.                     <id>scala-test-compile</id>
93.                     <phase>process-test-resources</phase>
94.                     <goals>
95.                         <goal>testCompile</goal>
96.                     </goals>
97.                 </execution>
98.             </executions>
99.         </plugin>
101.         <plugin>
102.             <groupId>org.apache.maven.plugins</groupId>
103.             <artifactId>maven-compiler-plugin</artifactId>
104.             <executions>
105.                 <execution>
106.                     <phase>compile</phase>
107.                     <goals>
108.                         <goal>compile</goal>
109.                     </goals>
110.                 </execution>
111.             </executions>
112.         </plugin>
114.         
115.         <plugin>
116.             <groupId>org.apache.maven.plugins</groupId>
117.             <artifactId>maven-shade-plugin</artifactId>
118.             <version>2.4.3</version>
119.             <executions>
120.                 <execution>
121.                     <phase>package</phase>
122.                     <goals>
123.                         <goal>shade</goal>
124.                     </goals>
125.                     <configuration>
126.                         <filters>
127.                             <filter>
128.                                 <artifact>*:*</artifact>
129.                                 <excludes>
130.                                     <exclude>META-INF/*.SF</exclude>
131.                                     <exclude>META-INF/*.DSA</exclude>
132.                                     <exclude>META-INF/*.RSA</exclude>
133.                                 </excludes>
134.                             </filter>
135.                         </filters>
136.                     </configuration>
137.                 </execution>
138.             </executions>
139.         </plugin>
140.     </plugins>
141. </build>

复制代码

创建一个scala目录

选择scala目录，右键，将目录转成源码包，或者点击maven的刷新按钮

编写Spark程序

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
4. /**
5.   * 1.创建SparkContext
6.   * 2.创建RDD
7.   * 3.调用RDD的Transformation（s）方法
8.   * 4.调用Action
9.   * 5.释放资源
10.   */
11. object WordCount {
13.   def main(args: Array[String]): Unit = {
15.     val conf: SparkConf = new SparkConf().setAppName("WordCount")
16.     //创建SparkContext，使用SparkContext来创建RDD
17.     val sc: SparkContext = new SparkContext(conf)
18.     //spark写Spark程序，就是对抽象的神奇的大集合【RDD】编程，调用它高度封装的API
19.     //使用SparkContext创建RDD
20.     val lines: RDD[String] = sc.textFile(args(0))
22.     //Transformation 开始 //
23.     //切分压平
24.     val words: RDD[String] = lines.flatMap(_.split(" "))
25.     //将单词和一组合放在元组中
26.     val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
27.     //分组聚合，reduceByKey可以先局部聚合再全局聚合
28.     val reduced: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)
29.     //排序
30.     val sorted: RDD[(String, Int)] = reduced.sortBy(_._2, false)
31.     //Transformation 结束 //
33.     //调用Action将计算结果保存到HDFS中
34.     sorted.saveAsTextFile(args(1))
35.     //释放资源
36.     sc.stop()
37.   }
38. }

复制代码

使用maven打包


提交任务

•        上传jar包到服务器，然后使用sparksubmit命令提交任务

1. /bigdata/spark-3.2.3-bin-hadoop3.2/bin/spark-submit \
2. --master spark://node-1.51doit.cn:7077 \
3. --executor-memory 1g --total-executor-cores 4 \
4. --class cn._51doit.spark.day01.WordCount \
5. /root/spark-in-action-1.0.jar hdfs://node-1.51doit.cn:9000/words.txt hdfs://node-1.51doit.cn:9000/out
7. 参数说明：
8. --master 指定masterd地址和端口，协议为spark://，端口是RPC的通信端口
9. --executor-memory 指定每一个executor的使用的内存大小
10. --total-executor-cores指定整个application总共使用了cores
11. --class 指定程序的main方法全类名
12. jar包路径 args0 args1

复制代码

Java编写Spark的WordCount

使用匿名实现类方式

1. import org.apache.spark.SparkConf;
2. import org.apache.spark.api.java.JavaPairRDD;
3. import org.apache.spark.api.java.JavaRDD;
4. import org.apache.spark.api.java.JavaSparkContext;
5. import org.apache.spark.api.java.function.FlatMapFunction;
6. import org.apache.spark.api.java.function.Function2;
7. import org.apache.spark.api.java.function.PairFunction;
8. import scala.Tuple2;
10. import java.util.Arrays;
11. import java.util.Iterator;
13. public class JavaWordCount {
15.     public static void main(String[] args) {
16.         SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
17.         //创建JavaSparkContext
18.         JavaSparkContext jsc = new JavaSparkContext(sparkConf);
19.         //使用JavaSparkContext创建RDD
20.         JavaRDD<String> lines = jsc.textFile(args[0]);
21.         //调用Transformation（s）
22.         //切分压平
23.         JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
24.             @Override
25.             public Iterator<String> call(String line) throws Exception {
26.                 return Arrays.asList(line.split(" ")).iterator();
27.             }
28.         });
29.         //将单词和一组合在一起
30.         JavaPairRDD<String, Integer> wordAndOne = words.mapToPair(
31.                 new PairFunction<String, String, Integer>() {
32.                     @Override
33.                     public Tuple2<String, Integer> call(String word) throws Exception {
34.                         return Tuple2.apply(word, 1);
35.                     }
36.         });
37.         //分组聚合
38.         JavaPairRDD<String, Integer> reduced = wordAndOne.reduceByKey(
39.                 new Function2<Integer, Integer, Integer>() {
40.             @Override
41.             public Integer call(Integer v1, Integer v2) throws Exception {
42.                 return v1 + v2;
43.             }
44.         });
45.         //排序，先调换KV的顺序VK
46.         JavaPairRDD<Integer, String> swapped = reduced.mapToPair(
47.                 new PairFunction<Tuple2<String, Integer>, Integer, String>() {
48.             @Override
49.             public Tuple2<Integer, String> call(Tuple2<String, Integer> tp) throws Exception {
50.                 return tp.swap();
51.             }
52.         });
53.         //再排序
54.         JavaPairRDD<Integer, String> sorted = swapped.sortByKey(false);
55.         //再调换顺序
56.         JavaPairRDD<String, Integer> result = sorted.mapToPair(
57.                 new PairFunction<Tuple2<Integer, String>, String, Integer>() {
58.             @Override
59.             public Tuple2<String, Integer> call(Tuple2<Integer, String> tp) throws Exception {
60.                 return tp.swap();
61.             }
62.         });
63.         //触发Action，将数据保存到HDFS
64.         result.saveAsTextFile(args[1]);
65.         //释放资源
66.         jsc.stop();
67.     }
68. }

复制代码

使用Lambda表达式方式

1. import org.apache.spark.SparkConf;
2. import org.apache.spark.api.java.JavaPairRDD;
3. import org.apache.spark.api.java.JavaRDD;
4. import org.apache.spark.api.java.JavaSparkContext;
5. import scala.Tuple2;
7. import java.util.Arrays;
9. public class JavaLambdaWordCount {
11.     public static void main(String[] args) {
12.         SparkConf conf = new SparkConf().setAppName("JavaLambdaWordCount");
13.         //创建SparkContext
14.         JavaSparkContext jsc = new JavaSparkContext(conf);
15.         //创建RDD
16.         JavaRDD<String> lines = jsc.textFile(args[0]);
17.         //切分压平
18.         JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
19.         //将单词和一组合
20.         JavaPairRDD<String, Integer> wordAndOne = words.mapToPair(word -> Tuple2.apply(word, 1));
21.         //分组聚合
22.         JavaPairRDD<String, Integer> reduced = wordAndOne.reduceByKey((a, b) -> a + b);
23.         //调换顺序
24.         JavaPairRDD<Integer, String> swapped = reduced.mapToPair(tp -> tp.swap());
25.         //排序
26.         JavaPairRDD<Integer, String> sorted = swapped.sortByKey(false);
27.         //调换顺序
28.         JavaPairRDD<String, Integer> result = sorted.mapToPair(tp -> tp.swap());
29.         //将数据保存到HDFS
30.         result.saveAsTextFile(args[1]);
31.         //释放资源
32.         jsc.stop();
33.     }
34. }

复制代码

本地运行Spark和Debug

spark程序每次都打包上在提交到集群上比较麻烦且不方便调试，Spark还可以进行Local模式运行，方便测试和调试
在本地运行

1. //Spark程序local模型运行,local[*]是本地运行，并开启多个线程
2. val conf: SparkConf = new SparkConf()
3.   .setAppName("WordCount")
4.   .setMaster("local[*]") //设置为local模式执行

复制代码

并输入运行参数
hdfs://linux01:9000/words.txt hdfs://linux01:9000/out/out01
读取HDFS中的数据

由于往HDFS中的写入数据存在权限问题，所以在代码中设置用户为HDFS目录的所属用户

1. //往HDFS中写入数据，将程序的所属用户设置成更HDFS一样的用户
2. System.setProperty("HADOOP_USER_NAME", "root")

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！