sparkRDD教程之基本命令

作者：nchu可乐百香果

指导者：nchu-YoungDragon

1.前期准备

（1）从迅雷网盘上面下载这个项目，并且把scala，maven和java环境配置好

网盘链接：

分享文件：SparkRDD.zip
链接：https://pan.xunlei.com/s/VOGUwt7hVp1ZQB6V7n1j6e70A1?pwd=q87r#
复制这段内容后打开迅雷，查看更方便

打开项目后，其目录结构如下：

（2）新建一个scala object文件 （点击新建scala文件后选择第三个选项object）

固定模版

.appName("task5"),这个名字建议改为自己当前的文件名

package com.itheimaimport org.apache.spark.sql.SparkSessionobject task5 {def main(args: Array[String]): Unit = {// 创建SparkSessionval spark = SparkSession.builder.appName("task5").master("local").getOrCreate()val sc = spark.sparkContext//在这里写代码spark.stop()}}

2.sparkRDD的基本命令

x._1表示元组的第一项（从1开始数这里，反正这样的写法是从1开始数的）

当然，x(0)是从0开始数的

（1）创建sparkRDD和打印输出

第一种方式

    val rdd1=sc.parallelize(List(1,2,3,4))println(rdd1.collect().mkString("\n"))

 结果

第二种方式

    val rdd2=sc.makeRDD(List(1,2,3,4))println(rdd2.collect().mkString("\n"))

 结果

第三种方式

    var score_path="src\\main\\resources\\score.txt"val rdd3=sc.textFile(score_path)println(rdd3.collect().mkString("\n"))

score.txt

student_id,course_code,score
108,3-105,99
105,3-105,88
107,3-105,77
105,3-245,87
108,3-245,89
107,3-245,82
106,3-245,74
107,6-101,75
108,6-101,82
106,6-101,65
109,6-102,99
101,6-102,79
105,9-106,81
106,9-106,97
107,9-106,65
108,9-106,100
109,9-106,82
105,6-102,85

（2）map命令

代码1 

    var rdd1=sc.parallelize(List(1,2,3,4))var res1=rdd1.map(x=>x*x*x)println(res1.collect().mkString("\n"))

结果 1

代码2

    var rdd2=sc.parallelize(List("a","b","c"))var res2=rdd2.map(x=>(x,1))println(res2.collect().mkString("\n"))

结果2

（3）flatmap命令（这个用起来有一些限制，所以使用的很少，我感觉不太好用，不如map简单直接）

代码 1

    val rdd2 = sc.parallelize(List("a b c", "d e f", "g h i"))val map_res = rdd2.map(x => x.split(" "))println(map_res.collect().mkString("\n"))val flatMap_res = rdd2.flatMap(x => x.split(" "))println(flatMap_res.collect().mkString("\n"))

结果1 

（4）sortBy命令（默认值是true，表示升序排序）

代码1

    val rdd = sc.parallelize(List('a','b','c','d','e'))val res = rdd.map(x =>(x,1)).sortBy(x=>x._1, true)//true可以省略println(res.collect().mkString("\n"))

截图1

代码2

    val rdd = sc.parallelize(List('a','b','c','d','e'))val res = rdd.map(x =>(x,1)).sortBy(x=>x._1, false)println(res.collect().mkString("\n"))

截图2

代码3

    val rdd = sc.parallelize(1 to 10)val res = rdd.map(x =>(x,1)).sortBy(x=>x._1, false)println(res.collect().mkString("\n"))

截图3

（5）filter命令（满足条件的留下来，和sql中的where命令类似）

代码1

    val rdd = sc.parallelize(1 to 10)val res = rdd.filter(x=>x<=5)println(res.collect().mkString("\n"))

截图1

代码2

    val rdd = sc.parallelize(1 to 10)val res = rdd.filter(x=>x>=3 && x<=5)println(res.collect().mkString("\n"))

截图2

（6）distinct命令（去重）

代码1

    val rdd = sc.parallelize(List(1,2,2,3,3))val res = rdd.distinct()println(res.collect().mkString("\n"))

截图1

（7）union命令（类似于求并集，但是不会自动去重）

代码1

    val rdd1 = sc.parallelize(List(1,2,3,4))val rdd2 = sc.parallelize(List(3,4,5,6))val res = rdd1.union(rdd2)println(res.collect().mkString("\n"))

截图1

（8）intersection（类似于求交集）

代码1

    val rdd1 = sc.parallelize(List(1,2,3,4))val rdd2 = sc.parallelize(List(3,4,5,6))val res = rdd1.intersection(rdd2)println(res.collect().mkString("\n"))

截图1

（9）subtract（类似于求差集）

代码1

    val rdd1 = sc.parallelize(List(1,2,3,4))val rdd2 = sc.parallelize(List(3,4,5,6))val res1_2 = rdd1.subtract(rdd2)println(res1_2.collect().mkString("\n"))

截图1

代码2

    val rdd1 = sc.parallelize(List(1,2,3,4))val rdd2 = sc.parallelize(List(3,4,5,6))val res2_1 = rdd2.subtract(rdd1)println(res2_1.collect().mkString("\n"))

截图2

（10）cartesion（求笛卡尔积）

代码1

    val rdd1 = sc.parallelize(List(1,2,3,4))val rdd2 = sc.parallelize(List(3,4,5,6))val res = rdd1.cartesian(rdd2)println(res.collect().mkString("\n"))

截图1

（11）keyBy命令

代码1-3的前期准备，初始化rdd

val rdd = sc.parallelize(List('a','b','c','d','e'))

代码1

    val res1 = rdd.map(x => (1, 2, x)).keyBy(x => x._1)println(res1.collect().mkString("\n"))

截图1

代码2

    val res2 = rdd.map(x => (1, 2, x)).keyBy(x => x._2)println(res2.collect().mkString("\n"))

截图2

代码3

    val res3 = rdd.map(x => (1, 2, x)).keyBy(x => x._3)println(res3.collect().mkString("\n"))

截图3

总结

    val rdd = sc.parallelize(List('a', 'b', 'c', 'd', 'e'))val res1 = rdd.map(x => (1, 2, x)).keyBy(x => x._1)val res11 = rdd.map(x => (1, 2, x)).map(x=>(x._1,x)) //这2行代码等价println(res1.collect().mkString("\n"))println(res11.collect().mkString("\n"))

（12）mapValues（对所有值进行相同的操作，同时要求是二元组结构）

代码1

    val rdd = sc.parallelize(List('a', 'b', 'c', 'd', 'e'))val rdd2=rdd.map(x=>(x,1))println(rdd2.collect().mkString("\n"))

截图1  （显然此时它们的value都为1）

代码2

    val rdd = sc.parallelize(List('a', 'b', 'c', 'd', 'e'))val rdd2=rdd.map(x=>(x,1)).mapValues(x=>x*20)println(rdd2.collect().mkString("\n"))

截图2

（13）reduceByKey（对相同键的值进行操作，同时要求是二元组结构）

代码1

    val rdd = sc.parallelize(List('a', 'b', 'c', 'd', 'e'))val rdd2=rdd.map(x=>(x,1)).mapValues(x=>x*20)var rdd4=rdd2.union(rdd2).union(rdd2)println(rdd4.collect().mkString("\n"))

截图1

代码2

    val rdd = sc.parallelize(List('a', 'b', 'c', 'd', 'e'))val rdd2=rdd.map(x=>(x,1)).mapValues(x=>x*20)var rdd4=rdd2.union(rdd2).union(rdd2)var rdd5=rdd4.reduceByKey((y1,y2)=>(y1+y2))println(rdd5.collect().mkString("\n"))

截图2

（14）groupBy命令（我感觉使用的不多，了解即可）

代码1

    val rdd = sc.parallelize(0 to 9)val rdd2=rdd.groupBy(x=>{if( x % 2==0) "even" else "odd" })println(rdd2.collect().mkString("\n"))

截图1

（15）groupByKey命令（我感觉使用的不多，了解即可）

代码1

    val rdd = sc.parallelize(List(('a',1),('b',2),('c',3)))val rdd1 = sc.parallelize(List(('a',4),('b',5),('c',6)))val rdd2=rdd.union(rdd1)println(rdd2.collect().mkString("\n"))

截图1

代码2

    val rdd = sc.parallelize(List(('a',1),('b',2),('c',3)))val rdd1 = sc.parallelize(List(('a',4),('b',5),('c',6)))val rdd2=rdd.union(rdd1).groupByKey()println(rdd2.collect().mkString("\n"))

截图2

（16）join命令

代码1

    val rdd1 = sc.parallelize(List(("K1", "V1"), ("K2", "V2"), ("K3", "V3")))val rdd2 = sc.parallelize(List(("K1", "V2"), ("K2", "V3"), ("K4", "V4")))var join1 = rdd1.join(rdd2)println(join1.collect().mkString("\n"))

截图1

代码2  （利用其他方法来实现和join类似的效果）

    val rdd1 = sc.parallelize(List(("K1", "V1"), ("K2", "V2"), ("K3", "V3")))val rdd2 = sc.parallelize(List(("K1", "V2"), ("K2", "V3"), ("K4", "V4")))var res2 = rdd1.union(rdd2).mapValues(x => (x, "V")).reduceByKey((y1, y2) => (y1._1, y2._1)).filter(x => x._2._2 != "V").sortBy(x => x._1)println(res2.collect().mkString("\n"))

截图2

（17）zip命令

代码1

    var rdd3=sc.makeRDD(1 to 5)var rdd4=sc.makeRDD(List('a','b','c','d','e'))var res3_4=rdd3.zip(rdd4)println(res3_4.collect().mkString("\n"))

截图1

代码2

    var rdd3 = sc.makeRDD(1 to 5)var rdd4 = sc.makeRDD(List('a', 'b', 'c', 'd', 'e'))var res4_3=rdd4.zip(rdd3)println(res4_3.collect().mkString("\n"))

截图2

（18）其他命令

3.总结一下

（1）第一点，也是重点

x._1表示元组的第一项（从1开始数这里，反正这样的写法是从1开始数的）

当然，x(0)是从0开始数的

count命令的返回值long类型，这一点要注意，经常需要使用toInt来转Int

（2）join命令的等价替换

这个其实主要还是为了方便自己更好理解这些命令

求点赞求收藏求关注

作者：nchu可乐百香果

指导者：nchu-YoungDragon

4.其他

（1）代码注释

package com.itheima

// 导入SparkSession和SaveMode等相关库
import org.apache.spark.sql.{SparkSession, SaveMode}

object task1 {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession对象，这个对象是操作Spark的入口
    val spark = SparkSession
      .builder
      .appName("WordCount")  // 设置应用程序名称
      .master("local")  // 设置运行模式为local，即在本地运行，可以通过集群模式连接远程集群
//      .master("spark://192.168.88.101:7077")  // 这行代码可以用于连接到远程Spark集群，当前是注释掉的
      .getOrCreate()  // 获取SparkSession对象，如果没有则创建

    // 获取SparkContext对象，SparkContext是Spark应用的核心，用来创建RDD
    val sc = spark.sparkContext

    // 读取HDFS上的两个文本文件并创建RDD
    val rdd1 = sc.textFile("hdfs://node1:8020/input/A.txt")  // 读取A.txt文件
    val rdd2 = sc.textFile("hdfs://node1:8020/input/B.txt")  // 读取B.txt文件

//    // 本地文件读取方式（注释掉）
//    val rdd1 = sc.textFile("D:\\新建文件夹 (3)\\2024大三上学期文件夹\\黑马Redis笔记\\SparkRDD\\src\\main\\java\\com\\springbootdemo\\A.txt")
//    val rdd2 = sc.textFile("D:\\新建文件夹 (3)\\2024大三上学期文件夹\\黑马Redis笔记\\SparkRDD\\src\\main\\java\\com\\springbootdemo\\B.txt")

    // 合并两个RDD，并进行处理：
    // 1. 使用union合并两个RDD
    // 2. 使用map将每行文本按空格分割，取第一个和第二个元素作为元组
    // 3. 使用sortBy按元组的第一个元素排序
    // 4. 使用distinct去重
    val rdd3 = rdd1.union(rdd2).map(x => (x.split(" ")(0), x.split(" ")(1))).sortBy(_._1).distinct()

    // 将RDD转换为DataFrame，DataFrame是SparkSQL中用于表示结构化数据的核心数据结构
    val data = spark.createDataFrame(rdd3).toDF("value1", "value2")  // 设置DataFrame的列名为value1和value2

    // 配置数据库连接参数，假设目标数据库为MySQL
    // 设置数据库的连接URL，包含主机地址和端口号
    // val jdbcUrl = "jdbc:mysql://192.168.88.101:3306/spark"  // 使用远程MySQL服务器的URL（注释掉）
    val jdbcUrl = "jdbc:mysql://127.0.0.1:3306/spark"  // 使用本地MySQL服务器的URL
    val jdbcUser = "root"  // 数据库的用户名
    val jdbcPassword = "123456"  // 数据库的密码

    // 将DataFrame写入MySQL数据库
    data.write
      .format("jdbc")  // 使用JDBC格式进行数据写入
      .option("url", jdbcUrl)  // 配置JDBC连接URL
      .option("dbtable", "spark_task1")  // 指定目标数据库表名
      .option("user", jdbcUser)  // 配置数据库用户名
      .option("password", jdbcPassword)  // 配置数据库密码
      .mode(SaveMode.Overwrite)  // 写入模式设置为覆盖（Overwrite），会覆盖已有数据
      .save()  // 执行保存操作

    // 停止SparkSession，释放资源
    spark.stop()
  }
}

代码流程简述：

1. 创建SparkSession：初始化Spark的核心环境，允许你使用SparkSQL等高级API。
2. 读取数据：从HDFS或本地文件系统读取两个文本文件，创建RDD。
3. 数据转换：对读取的RDD进行转换，进行数据的合并、分割、排序和去重操作。
4. 转换为DataFrame：将RDD转换为DataFrame以便使用SQL等功能。
5. 数据库配置：设置MySQL数据库连接的URL、用户名和密码。
6. 将数据写入数据库：将处理后的DataFrame写入到指定的MySQL表中。
7. 关闭SparkSession：释放资源，结束Spark应用。

（2）什么是SparkSession

`SparkSession` 是 Spark 2.0 引入的一个统一入口点，它简化了与 Spark 相关的所有操作，包括批处理、流处理、机器学习和图计算等。

在 Spark 1.x 中，用户使用 `SQLContext` 和 `HiveContext` 等不同的上下文来操作 Spark SQL 数据库或者使用 RDD API，而 `SparkSession` 将这些功能整合在一起，成为一个统一的入口，使得用户不需要区分具体的上下文。

### 主要功能和作用：

1. **创建 DataFrame 和 DataSet**：
   `SparkSession` 允许用户通过 `read` 接口创建 DataFrame 或者 DataSet，支持多种数据源如：CSV、JSON、Parquet、JDBC、HDFS 等。
   
2. **访问 Spark SQL**：
   通过 `SparkSession` 可以直接执行 SQL 查询，执行结果以 DataFrame 形式返回。

3. **Spark 应用的入口点**：
   `SparkSession` 提供了访问 `SparkContext`、`SQLContext`、`HiveContext` 等功能，因此它是操作 Spark 集群的主入口。

4. **管理 Spark 配置**：
   `SparkSession` 允许用户设置 Spark 配置参数，控制 Spark 作业的行为，比如内存大小、执行模式等。

5. **集成 Hive**：
   在 Spark 2.0 中，`SparkSession` 也提供了对 Hive 的支持，使得 Spark 可以无缝集成到 Hive 环境中，支持执行 HiveQL 查询。

### 创建 `SparkSession` 的基本步骤：
1. **初始化 SparkSession**：
   使用 `SparkSession.builder()` 创建一个 SparkSession 实例，并配置应用名称、运行模式等。

   ```scala
   val spark = SparkSession
     .builder()
     .appName("Example Application")
     .master("local[*]")  // 运行模式（local表示在本地运行，*表示使用所有CPU核心）
     .getOrCreate()  // 获取或创建 SparkSession
   ```

2. **通过 SparkSession 执行 SQL 操作**：
   ```scala
   val df = spark.read.json("path_to_json_file")
   df.createOrReplaceTempView("table_name")  // 创建临时视图
   val result = spark.sql("SELECT * FROM table_name")  // 执行SQL查询
   result.show()
   ```

3. **通过 SparkSession 访问 SparkContext**：
   `SparkSession` 提供了对 `SparkContext` 的访问，可以通过 `spark.sparkContext` 获取。
   ```scala
   val sc = spark.sparkContext  // 获取 SparkContext
   ```

### SparkSession 的重要属性和方法：
- `spark.read`：用于读取数据（如 CSV、JSON、Parquet 等）。
- `spark.sql()`：用于执行 SQL 查询。
- `spark.catalog`：用于管理表、视图等元数据。
- `spark.udf`：用于注册用户自定义函数（UDF）。
- `spark.version`：用于获取 Spark 版本。

### 示例：
```scala
val spark = SparkSession.builder()
  .appName("Spark Session Example")
  .master("local")
  .getOrCreate()

// 使用 SparkSession 读取数据
val df = spark.read.csv("path/to/data.csv")

// 执行 SQL 查询
df.createOrReplaceTempView("my_table")
val result = spark.sql("SELECT * FROM my_table WHERE column1 > 100")
result.show()

// 停止 SparkSession
spark.stop()
```

通过 `SparkSession`，Spark 提供了一个统一的接口来执行不同类型的操作，简化了 Spark 程序的编写和执行过程。