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

标题: 大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算效果 [打印本页]
作者: 石小疯    时间: 2024-8-22 09:16
标题: 大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算效果
点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：


章节内容

上节完成的内容如下：


Super Word Count

需求背景


编写代码

先实现到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 { … }


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


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


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


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


    val sc = new SparkContext(conf)


    sc.setLogLevel(“WARN”)


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


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


    map(_.toLowerCase)


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


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


    map((_, 1))


    reduceByKey(_ + _)


    sortBy(_._2, false)


    collect().foreach(println)


    sc.stop()


    添加依赖

    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
    留意：

    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企服之家，中国第一个企服评测及商务社交产业平台。



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