大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,高可用系列 2核4GB
简介: 大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL

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

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka(已更完)

Spark(正在更新!)

章节内容

上节完成的内容如下:


Spark案例编写 Scala

计算圆周率

找共同的好友

Super Word Count

需求背景

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

编写代码

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

package icu.wzk

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

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))
    lines
      .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()
  }

}

详细解释

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,释放资源。

添加依赖

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.28</version>
</dependency>

同时我们需要在build的部分,也要加入对应的内容,让驱动可以加载进来:

<build>
    <plugins>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>4.4.0</version>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.3.0</version>
            <configuration>
                <archive>
                    <manifest>
                        <mainClass>cn.lagou.sparkcore.WordCount</mainClass>
                    </manifest>
                </archive>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

创建库表

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


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

写入SQL-未优化

我们在 foreach 中保存了数据,此时需要创建大量的MySQL连接,效率是比较低的。

package icu.wzk

import com.mysql.cj.xdevapi.PreparableStatement
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object SuperWordCount2 {

  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("ScalaSuperWordCount2")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

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

    val words: RDD[String] = lines
      .flatMap(_.split("\\s+"))
      .map(_.trim.toLowerCase())

    val clearWords: RDD[String] = words
      .filter(!stopWords.contains(_))
      .map(_.replaceAll(punctuation, ""))

    val result: RDD[(String, Int)] = clearWords
      .map((_, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
    result.foreach(println)

    // 输出到 MySQL
    val username = "hive"
    val password = "hive@wzk.icu"
    val url = "jdbc:mysql://h122.wzk.icu:3306/spark-test?useUnicode=true&characterEncoding=utf-8&useSSL=false"

    var conn: Connection = null
    var stmt: PreparedStatement = null
    var sql = "insert into wordcount values(?, ?)"

    result.foreach{
      case (word, count) => try {
        conn = DriverManager.getConnection(url, username, password)
        stmt = conn.prepareStatement(sql)
        stmt.setString(1, word)
        stmt.setInt(2, count)
      } catch {
        case e: Exception => e.printStackTrace()
      } finally {
        if (stmt != null) {
          stmt.close()
        }
        if (conn != null) {
          conn.close()
        }
      }
    }

    sc.stop()
  }

}

写入SQL-优化版

优化后使用 foreachPartition 保存数据,一个分区创建一个链接:cache RDD

注意:

  • SparkSQL 有方便的读写MySQL的方法,给参数直接调用即可
  • 但掌握这个方法很重要,因为SparkSQL不是支持所有类型的数据库
package icu.wzk

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object SuperWordCount3 {

  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 = "[\\)\\.,:;'!\\?]"
  private val username = "hive"
  private val password = "hive@wzk.icu"
  private val url = "jdbc:mysql://h122.wzk.icu:3306/spark-test?useUnicode=true&characterEncoding=utf-8&useSSL=false"

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("ScalaSuperWordCount2")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

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

    val words: RDD[String] = lines
      .flatMap(_.split("\\s+"))
      .map(_.trim.toLowerCase())

    val clearWords: RDD[String] = words
      .filter(!stopWords.contains(_))
      .map(_.replaceAll(punctuation, ""))

    val result: RDD[(String, Int)] = clearWords
      .map((_, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
    result.foreach(println)

    result.foreachPartition(saveAsMySQL)

    sc.stop()
  }

  def saveAsMySQL(iter: Iterator[(String, Int)]): Unit = {
    var conn: Connection = null
    var stmt: PreparedStatement = null
    var sql = "insert into wordcount values(?, ?)"

    try {
      conn = DriverManager.getConnection(url, username, password)
      stmt = conn.prepareStatement(sql)
      iter.foreach{
        case (word, count) =>
          stmt.setString(1, word)
          stmt.setInt(2, count)
      }
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      if (stmt != null) {
        stmt.close()
      }
      if (conn != null) {
        conn.close()
      }
    }
  }

}

打包上传

mvn clean package

打包并上传到项目:

运行项目

不写入SQL版

不写入SQL版

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

运行结果如下图:

写入SQL-未优化版

spark-submit --master local[*] --class icu.wzk.SuperWordCount2 spark-wordcount-1.0-SNA

写入SQL-优化版

spark-submit --master local[*] --class icu.wzk.SuperWordCount3 spark-wordcount-1.0-SN

运行结果如下图:

查看数据

查看数据库,内容如下:

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
23天前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
31 0
|
23天前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
64 0
|
22天前
|
Java 大数据 数据库连接
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
24 2
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
|
18天前
|
关系型数据库 MySQL Java
Django学习二:配置mysql,创建model实例,自动创建数据库表,对mysql数据库表已经创建好的进行直接操作和实验。
这篇文章是关于如何使用Django框架配置MySQL数据库,创建模型实例,并自动或手动创建数据库表,以及对这些表进行操作的详细教程。
48 0
Django学习二:配置mysql,创建model实例,自动创建数据库表,对mysql数据库表已经创建好的进行直接操作和实验。
|
19天前
|
Java 关系型数据库 MySQL
springboot学习五:springboot整合Mybatis 连接 mysql数据库
这篇文章是关于如何使用Spring Boot整合MyBatis来连接MySQL数据库,并进行基本的增删改查操作的教程。
31 0
springboot学习五:springboot整合Mybatis 连接 mysql数据库
|
19天前
|
Java 关系型数据库 MySQL
springboot学习四:springboot链接mysql数据库,使用JdbcTemplate 操作mysql
这篇文章是关于如何使用Spring Boot框架通过JdbcTemplate操作MySQL数据库的教程。
17 0
springboot学习四:springboot链接mysql数据库,使用JdbcTemplate 操作mysql
|
23天前
|
消息中间件 存储 druid
大数据-156 Apache Druid 案例实战 Scala Kafka 订单统计
大数据-156 Apache Druid 案例实战 Scala Kafka 订单统计
32 3
|
22天前
|
关系型数据库 MySQL 数据库
mysql关系型数据库的学习
mysql关系型数据库的学习
16 0
|
23天前
|
SQL 分布式计算 大数据
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(一)
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(一)
28 0
|
23天前
|
SQL 分布式计算 大数据
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(二)
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(二)
25 0