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.itheima

import org.apache.spark.sql.SparkSession


object task5 {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val 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()
}
}

代码流程简述：

创建SparkSession：初始化Spark的核心环境，允许你使用SparkSQL等高级API。
读取数据：从HDFS或本地文件系统读取两个文本文件，创建RDD。
数据转换：对读取的RDD进行转换，进行数据的合并、分割、排序和去重操作。
转换为DataFrame：将RDD转换为DataFrame以便使用SQL等功能。
数据库配置：设置MySQL数据库连接的URL、用户名和密码。
将数据写入数据库：将处理后的DataFrame写入到指定的MySQL表中。
关闭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 程序的编写和执行过程。