ToB企服应用市场:ToB评测及商务社交产业平台

标题: Spark编程基础 [打印本页]
作者: 十念    时间: 2023-6-28 21:03
标题: 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")
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4