devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

IDEA编写Spark案例WordCount(Scala和Java编程)_idea 2023开发spark demo

作者：菜鸟追梦旅行 | 2024-04-28 07:31:16

赞

踩

idea 2023开发spark demo

用idea编写Spark程序

创建RDD,然后对RDD进行操作（调用RDD的方法，方法分为两类，一类叫Transformation（懒，lazy）,一类叫Action(执行程序)）

RDD上的方法和Scala原生的方法是有区别的

写好程序，打包上集群运行

本地模式运行spark程序，.setMaster("local[*]")

1.Scala编写

1.1 配置pom.xml文件

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.  
  7.     <groupId>com.caimh.spark</groupId>
  8.     <artifactId>sparktest</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.  
  11.     <properties>
  12.         <maven.compiler.source>1.8</maven.compiler.source>
  13.         <maven.compiler.target>1.8</maven.compiler.target>
  14.         <scala.version>2.11.8</scala.version>
  15.         <spark.version>2.2.0</spark.version>
  16.         <hadoop.version>2.6.5</hadoop.version>
  17.         <encoding>UTF-8</encoding>
  18.     </properties>
  19.  
  20.     <dependencies>
  21.         <!-- 导入scala的依赖 -->
  22.         <dependency>
  23.             <groupId>org.scala-lang</groupId>
  24.             <artifactId>scala-library</artifactId>
  25.             <version>${scala.version}</version>
  26.         </dependency>
  27.  
  28.         <!-- 导入spark的依赖 -->
  29.         <dependency>
  30.             <groupId>org.apache.spark</groupId>
  31.             <artifactId>spark-core_2.11</artifactId>
  32.             <version>${spark.version}</version>
  33.         </dependency>
  34.  
  35.         <!-- 指定hadoop-client API的版本 -->
  36.         <dependency>
  37.             <groupId>org.apache.hadoop</groupId>
  38.             <artifactId>hadoop-client</artifactId>
  39.             <version>${hadoop.version}</version>
  40.         </dependency>
  41.  
  42.     </dependencies>
  43.  
  44.     <build>
  45.         <pluginManagement>
  46.             <plugins>
  47.                 <!-- 编译scala的插件 -->
  48.                 <plugin>
  49.                     <groupId>net.alchim31.maven</groupId>
  50.                     <artifactId>scala-maven-plugin</artifactId>
  51.                     <version>3.2.2</version>
  52.                 </plugin>
  53.                 <!-- 编译java的插件 -->
  54.                 <plugin>
  55.                     <groupId>org.apache.maven.plugins</groupId>
  56.                     <artifactId>maven-compiler-plugin</artifactId>
  57.                     <version>3.5.1</version>
  58.                 </plugin>
  59.             </plugins>
  60.         </pluginManagement>
  61.         <plugins>
  62.             <plugin>
  63.                 <groupId>net.alchim31.maven</groupId>
  64.                 <artifactId>scala-maven-plugin</artifactId>
  65.                 <executions>
  66.                     <execution>
  67.                         <id>scala-compile-first</id>
  68.                         <phase>process-resources</phase>
  69.                         <goals>
  70.                             <goal>add-source</goal>
  71.                             <goal>compile</goal>
  72.                         </goals>
  73.                     </execution>
  74.                     <execution>
  75.                         <id>scala-test-compile</id>
  76.                         <phase>process-test-resources</phase>
  77.                         <goals>
  78.                             <goal>testCompile</goal>
  79.                         </goals>
  80.                     </execution>
  81.                 </executions>
  82.             </plugin>
  83.  
  84.             <plugin>
  85.                 <groupId>org.apache.maven.plugins</groupId>
  86.                 <artifactId>maven-compiler-plugin</artifactId>
  87.                 <executions>
  88.                     <execution>
  89.                         <phase>compile</phase>
  90.                         <goals>
  91.                             <goal>compile</goal>
  92.                         </goals>
  93.                     </execution>
  94.                 </executions>
  95.             </plugin>
  96.  
  97.             <!-- 打jar插件 -->
  98.             <plugin>
  99.                 <groupId>org.apache.maven.plugins</groupId>
  100.                 <artifactId>maven-shade-plugin</artifactId>
  101.                 <version>2.4.3</version>
  102.                 <executions>
  103.                     <execution>
  104.                         <phase>package</phase>
  105.                         <goals>
  106.                             <goal>shade</goal>
  107.                         </goals>
  108.                         <configuration>
  109.                             <filters>
  110.                                 <filter>
  111.                                     <artifact>*:*</artifact>
  112.                                     <excludes>
  113.                                         <exclude>META-INF/*.SF</exclude>
  114.                                         <exclude>META-INF/*.DSA</exclude>
  115.                                         <exclude>META-INF/*.RSA</exclude>
  116.                                     </excludes>
  117.                                 </filter>
  118.                             </filters>
  119.                         </configuration>
  120.                     </execution>
  121.                 </executions>
  122.             </plugin>
  123.         </plugins>
  124.     </build>
  125. </project>

1.2 编写spark程序

  1. package com.caimh.spark
  2.  
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.{SparkConf, SparkContext}
  5.  
  6. /**
  7.   * Created by caimh on 2019/10/28.
  8.   */
  9. object ScalaWordCount {
  10.  
  11.   def main(args: Array[String]): Unit = {
  12.     //1.创建spark配置，设置应用程序名字
  13.     val conf = new SparkConf().setAppName("ScalaWordCount")
  14.     //2.创建spark执行的入口
  15.     val sc = new SparkContext(conf)
  16.     //3.指定以后从哪里读取数据创建RDD(弹性分布式数据集)
  17.     val lines: RDD[String] = sc.textFile(args(0))
  18.     //切分压平
  19.     val words: RDD[String] = lines.flatMap(_.split(" "))
  20.     //将单词和1组合
  21.     val wordmap: RDD[(String, Int)] = words.map((_, 1))
  22.     //按key进行聚合
  23.     val wordreduced: RDD[(String, Int)] = wordmap.reduceByKey(_ + _)
  24.     //排序
  25.     val wordsorted: RDD[(String, Int)] = wordreduced.sortBy(_._2, false)
  26.     //将结果保存到HDFS中
  27.     wordsorted.saveAsTextFile(args(1))
  28.     //释放资源
  29.     sc.stop()
  30.   }
  31.  
  32. }

1.3 使用Maven打包

1.4 选择编译成功的jar包，并将该jar上传到Spark集群中的某个节点上

1.5 首先启动集群

如果是Spark配置的是Standalone集群，需要启动spark集群，hdfs集群

如果配置的是Yarn集群，启动hdfs/yarn集群。

我是配置的yarn集群

  1. [caimh@master-node spark-2.1.1]$ jps
  2. 46368 Jps
  3. 44469 ResourceManager
  4. 45048 QuorumPeerMain
  5. 44056 NameNode
  6. 44585 NodeManager
  7. 44170 DataNode

1.6 使用spark-submit命令提交Spark应用（注意参数的顺序）

  1. [caimh@master-node spark-2.1.1]$ bin/spark-submit \
  2. > --master yarn-cluster \
  3. > --class com.caimh.spark.ScalaWordCount \
  4. > /opt/module/spark-2.1.1/sparktest-1.0-SNAPSHOT.jar hdfs://master-node:9000/wordcount.txt hdfs://master-node:9000/wordcount

1.7 查看程序执行结果

yarn查看

hdfs查看

2.Java编写

 

1.2 编写spark程序

  1. package com.caimh.spark;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.FlatMapFunction;
  8. import org.apache.spark.api.java.function.Function2;
  9. import org.apache.spark.api.java.function.PairFunction;
  10. import scala.Tuple2;
  11.  
  12. import java.util.Arrays;
  13. import java.util.Iterator;
  14.  
  15. /**
  16.  * Created by caimh on 2019/10/29.
  17.  */
  18. public class JavaWordCount {
  19.  
  20.     public static void main(String[] args) {
  21.  
  22.         //创建配置文件
  23.         SparkConf conf = new SparkConf().setAppName("JavaWordCount");
  24.         //创建SparkContext
  25.         JavaSparkContext jsc = new JavaSparkContext(conf);
  26.         //指定以后从哪里读取数据
  27.         JavaRDD<String> lines = jsc.textFile(args[0]);
  28.         //切分压平(函数式编程)
  29.         JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
  30.             @Override
  31.             public Iterator<String> call(String line) throws Exception {
  32.                 return Arrays.asList(line.split(" ")).iterator();
  33.             }
  34.         });
  35.         //将单词和1组合在一起
  36.         JavaPairRDD<String, Integer> wordTuple = words.mapToPair(new PairFunction<String, String, Integer>() {
  37.             @Override
  38.             public Tuple2<String, Integer> call(String word) throws Exception {
  39.  
  40.                 return new Tuple2<>(word, 1);
  41.             }
  42.         });
  43.         //根据key聚合
  44.         JavaPairRDD<String, Integer> wordReduced = wordTuple.reduceByKey(new Function2<Integer, Integer, Integer>() {
  45.             @Override
  46.             public Integer call(Integer v1, Integer v2) throws Exception {
  47.                 return v1 + v2;
  48.             }
  49.         });
  50.         //排序(方法sordBykey是根据key排序，需要把wordReduced的k,v顺序调换)
  51.         //调换顺序
  52.         JavaPairRDD<Integer, String> wordSwaped = wordReduced.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
  53.             @Override
  54.             public Tuple2<Integer, String> call(Tuple2<String, Integer> tp) throws Exception {
  55.                 return tp.swap();
  56.             }
  57.         });
  58.         //排序
  59.         wordSwaped.sortByKey(false);
  60.         //调换顺序
  61.         JavaPairRDD<String, Integer> result = wordSwaped.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
  62.             @Override
  63.             public Tuple2<String, Integer> call(Tuple2<Integer, String> tp) throws Exception {
  64.                 return tp.swap();
  65.             }
  66.         });
  67.         //将数据保存到HDFS
  68.         result.saveAsTextFile(args[1]);
  69.         //释放资源
  70.         jsc.close();
  71.     }
  72. }

1.3 编译打包

1.4 执行程序

  1. [caimh@master-node spark-2.1.1]$ bin/spark-submit 
  2. --master yarn-cluster 
  3. --class com.caimh.spark.JavaWordCount 
  4. /opt/module/spark-2.1.1/sparktest-1.0-SNAPSHOT.jar hdfs://master-node:9000/wordcount.txt hdfs://master-node:9000/wordcount

1.4 检查结果

yarn

hdfs

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/501068
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号