大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算效果 ...

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
  
• HDFS（已更完）
  
• MapReduce（已更完）
  
• Hive（已更完）
  
• Flume（已更完）
  
• Sqoop（已更完）
  
• Zookeeper（已更完）
  
• HBase（已更完）
  
• Redis （已更完）
  
• Kafka（已更完）
  
• Spark（正在更新！）
  

章节内容

上节完成的内容如下：

• Spark案例编写 Scala
  
• 计算圆周率
  
• 找共同的好友
  

Super Word Count

需求背景

• 给定一段文本
  
• 将单词全部转换为小写
  
• 去除标点符号
  
• 去除停用词
  
• count值降序生存
  
• 效果生存到MySQL
  
• 额外要求：标点符合和停用词可以自界说
  

编写代码

先实现到MySQL生存前的内容，我们须要先编写测试一下我们的代码是否正确

1. package icu.wzk
3. import org.apache.spark.rdd.RDD
4. import org.apache.spark.{SparkConf, SparkContext}
6. object SuperWordCount1 {
8.   private val stopWords = "in on to from by a an the is are were was i we you your he his".split("\\s+")
10.   private val punctuation = "[\\)\\.,:;'!\\?]"
12.   def main(args: Array[String]): Unit = {
13.     val conf = new SparkConf()
14.       .setAppName("ScalaSuperWordCount1")
15.       .setMaster("local[*]")
16.     val sc = new SparkContext(conf)
17.     sc.setLogLevel("WARN")
19.     val lines: RDD[String] = sc.textFile(args(0))
20.     lines
21.       .flatMap(_.split("\\s+"))
22.       .map(_.toLowerCase)
23.       .map(_.replaceAll(punctuation, ""))
24.       .filter(word => !stopWords.contains(word) && word.trim.nonEmpty)
25.       .map((_, 1))
26.       .reduceByKey(_ + _)
27.       .sortBy(_._2, false)
28.       .collect()
29.       .foreach(println)
30.    
31.     sc.stop()
32.   }
34. }

复制代码

详细表明

object SuperWordCount1 { … }

• SuperWordCount1 是一个 Scala 对象，界说了一个单例对象用于运行单词计数步伐。
  

private val stopWords = “in on to from by a an the is are were was i we you your he his”.split(“\s+”)

• 这里界说了一个 stopWords 列表，包罗了常见的停用词，这些词在统计单词频率时会被过滤掉。
  
• split(“\s+”) 方法将这些停用词用空白字符分割成数组，便于后续的查找和过滤。
  

private val punctuation = “[\)\.,:;'!\?]”

• 界说了一个正则表达式 punctuation，用于匹配常见的标点符号。这些标点符号在统计单词频率时会被去除。
  

def main(args: Array[String]): Unit = { … }

• main 方法是步伐的入口点，args 是下令行参数，其中 args(0) 通常表示输入文件的路径。
  

val conf = new SparkConf().setAppName(“ScalaSuperWordCount1”).setMaster(“local

”)

• SparkConf() 用于配置 Spark 应用步伐。setAppName(“ScalaSuperWordCount1”) 设置应用步伐的名称。
  
• setMaster(“local
  
• ”) 指定应用步伐以本地模式运行，使用全部可用的 CPU 核心。
  

val sc = new SparkContext(conf)

• SparkContext 是 Spark 应用步伐的核心，用于与 Spark 集群进行交互。
  

sc.setLogLevel(“WARN”)

• 设置日记级别为 “WARN”，淘汰日记输出，方便检察紧张信息。
  

val lines: RDD[String] = sc.textFile(args(0))

• sc.textFile(args(0)) 从指定的文本文件路径加载数据，创建一个 RDD[String]，其中每一行文本都作为一个字符串元素。
  
• lines 是包罗输入文本数据的 RDD。
  

flatMap(_.split(“\s+”))

• flatMap 方法将每一行字符串按空白字符拆分成单词，并将其展开成单个单词的 RDD。
  

map(_.toLowerCase)

• 将每个单词转换为小写，以确保统计时不区分巨细写。
  

map(_.replaceAll(punctuation, “”))

• 使用正则表达式 punctuation 去除单词中的标点符号，使得统计效果更加正确。
  

filter(word => !stopWords.contains(word) && word.trim.nonEmpty)

• filter 方法过滤掉停用词和空白单词：
  
• !stopWords.contains(word) 确保当前单词不在停用词列表中。
  
• word.trim.nonEmpty 确保单词在去除前后空白字符后不是空字符串。
  

map((_, 1))

• 将每个单词映射为 (word, 1) 的键值对，表示每个单词出现一次。
  

reduceByKey(_ + _)

• reduceByKey 方法根据键（单词）对值（计数）进行累加，统计每个单词的总出现次数。
  

sortBy(_._2, false)

• 将统计效果按值（单词出现的次数）从大到小排序。
  

collect().foreach(println)

• collect() 方法将 RDD 中的数据网络到驱动步伐中（即本地），然后使用 foreach(println) 输出每个单词及其出现的次数。
  
• 由于 collect 会将数据从分布式情况中拉到本地，须要留意数据量大的情况下大概导致内存不敷的题目。
  

sc.stop()

• 在计算完成后，调用 sc.stop() 方法停止 SparkContext，释放资源。
  

添加依赖

1. <dependency>
2.     <groupId>mysql</groupId>
3.     <artifactId>mysql-connector-java</artifactId>
4.     <version>8.0.28</version>
5. </dependency>

复制代码

同时我们须要在build的部门，也要加入对应的内容，让驱动可以加载进来：

1. <build>
2.     <plugins>
3.         <plugin>
4.             <groupId>net.alchim31.maven</groupId>
5.             <artifactId>scala-maven-plugin</artifactId>
6.             <version>4.4.0</version>
7.             <executions>
8.                 <execution>
9.                     <goals>
10.                         <goal>compile</goal>
11.                         <goal>testCompile</goal>
12.                     </goals>
13.                 </execution>
14.             </executions>
15.         </plugin>
16.         <plugin>
17.             <groupId>org.apache.maven.plugins</groupId>
18.             <artifactId>maven-assembly-plugin</artifactId>
19.             <version>3.3.0</version>
20.             <configuration>
21.                 <archive>
22.                     <manifest>
23.                         <mainClass>cn.lagou.sparkcore.WordCount</mainClass>
24.                     </manifest>
25.                 </archive>
26.                 <descriptorRefs>
27.                     <descriptorRef>jar-with-dependencies</descriptorRef>
28.                 </descriptorRefs>
29.             </configuration>
30.             <executions>
31.                 <execution>
32.                     <phase>package</phase>
33.                     <goals>
34.                         <goal>single</goal>
35.                     </goals>
36.                 </execution>
37.             </executions>
38.         </plugin>
39.     </plugins>
40. </build>

复制代码

创建库表

我们新建一个数据库，也要新建一个数据表

1. CREATE TABLE `wordcount` (
2.   `word` varchar(255) DEFAULT NULL,
3.   `count` int(11) DEFAULT NULL
4. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

复制代码

写入SQL-未优化

我们在 foreach 中生存了数据，此时须要创建大量的MySQL毗连，服从是比力低的。

1. package icu.wzk
3. import com.mysql.cj.xdevapi.PreparableStatement
4. import org.apache.spark.rdd.RDD
5. import org.apache.spark.{SparkConf, SparkContext}
7. import java.sql.{Connection, DriverManager, PreparedStatement}
9. object SuperWordCount2 {
11.   private val stopWords = "in on to from by a an the is are were was i we you your he his".split("\\s+")
13.   private val punctuation = "[\\)\\.,:;'!\\?]"
15.   def main(args: Array[String]): Unit = {
16.     val conf = new SparkConf()
17.       .setAppName("ScalaSuperWordCount2")
18.       .setMaster("local[*]")
19.     val sc = new SparkContext(conf)
20.     sc.setLogLevel("WARN")
22.     val lines: RDD[String] = sc.textFile(args(0))
24.     val words: RDD[String] = lines
25.       .flatMap(_.split("\\s+"))
26.       .map(_.trim.toLowerCase())
28.     val clearWords: RDD[String] = words
29.       .filter(!stopWords.contains(_))
30.       .map(_.replaceAll(punctuation, ""))
32.     val result: RDD[(String, Int)] = clearWords
33.       .map((_, 1))
34.       .reduceByKey(_ + _)
35.       .sortBy(_._2, false)
36.     result.foreach(println)
38.     // 输出到 MySQL
39.     val username = "hive"
40.     val password = "hive@wzk.icu"
41.     val url = "jdbc:mysql://h122.wzk.icu:3306/spark-test?useUnicode=true&characterEncoding=utf-8&useSSL=false"
43.     var conn: Connection = null
44.     var stmt: PreparedStatement = null
45.     var sql = "insert into wordcount values(?, ?)"
47.     result.foreach{
48.       case (word, count) => try {
49.         conn = DriverManager.getConnection(url, username, password)
50.         stmt = conn.prepareStatement(sql)
51.         stmt.setString(1, word)
52.         stmt.setInt(2, count)
53.       } catch {
54.         case e: Exception => e.printStackTrace()
55.       } finally {
56.         if (stmt != null) {
57.           stmt.close()
58.         }
59.         if (conn != null) {
60.           conn.close()
61.         }
62.       }
63.     }
65.     sc.stop()
66.   }
68. }

复制代码

写入SQL-优化版

优化后使用 foreachPartition 生存数据，一个分区创建一个链接：cache RDD
留意：

• SparkSQL 有方便的读写MySQL的方法，给参数直接调用即可
  
• 但把握这个方法很紧张，因为SparkSQL不是支持全部类型的数据库
  

1. package icu.wzk
3. import org.apache.spark.rdd.RDD
4. import org.apache.spark.{SparkConf, SparkContext}
6. import java.sql.{Connection, DriverManager, PreparedStatement}
8. object SuperWordCount3 {
10.   private val stopWords = "in on to from by a an the is are were was i we you your he his".split("\\s+")
11.   private val punctuation = "[\\)\\.,:;'!\\?]"
12.   private val username = "hive"
13.   private val password = "hive@wzk.icu"
14.   private val url = "jdbc:mysql://h122.wzk.icu:3306/spark-test?useUnicode=true&characterEncoding=utf-8&useSSL=false"
16.   def main(args: Array[String]): Unit = {
17.     val conf = new SparkConf()
18.       .setAppName("ScalaSuperWordCount2")
19.       .setMaster("local[*]")
20.     val sc = new SparkContext(conf)
21.     sc.setLogLevel("WARN")
23.     val lines: RDD[String] = sc.textFile(args(0))
25.     val words: RDD[String] = lines
26.       .flatMap(_.split("\\s+"))
27.       .map(_.trim.toLowerCase())
29.     val clearWords: RDD[String] = words
30.       .filter(!stopWords.contains(_))
31.       .map(_.replaceAll(punctuation, ""))
33.     val result: RDD[(String, Int)] = clearWords
34.       .map((_, 1))
35.       .reduceByKey(_ + _)
36.       .sortBy(_._2, false)
37.     result.foreach(println)
39.     result.foreachPartition(saveAsMySQL)
41.     sc.stop()
42.   }
44.   def saveAsMySQL(iter: Iterator[(String, Int)]): Unit = {
45.     var conn: Connection = null
46.     var stmt: PreparedStatement = null
47.     var sql = "insert into wordcount values(?, ?)"
49.     try {
50.       conn = DriverManager.getConnection(url, username, password)
51.       stmt = conn.prepareStatement(sql)
52.       iter.foreach{
53.         case (word, count) =>
54.           stmt.setString(1, word)
55.           stmt.setInt(2, count)
56.       }
57.     } catch {
58.       case e: Exception => e.printStackTrace()
59.     } finally {
60.       if (stmt != null) {
61.         stmt.close()
62.       }
63.       if (conn != null) {
64.         conn.close()
65.       }
66.     }
67.   }
69. }

复制代码

打包上传

1. mvn clean package

复制代码

打包并上传到项目：

运行项目

不写入SQL版

1. spark-submit --master local[*] --class icu.wzk.SuperWordCount1 spark-wordcount-1.0-SNAPSHOT.jar /opt/wzk/goodtbl.java

复制代码

运行效果如下图：

写入SQL-未优化版

1. spark-submit --master local[*] --class icu.wzk.SuperWordCount2 spark-wordcount-1.0-SNAPSHOT.jar /opt/wzk/goodtbl.java

复制代码

写入SQL-优化版

1. spark-submit --master local[*] --class icu.wzk.SuperWordCount3 spark-wordcount-1.0-SNAPSHOT.jar /opt/wzk/goodtbl.java

复制代码

运行效果如下图：

检察数据

检察数据库，内容如下：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。