Spark SQL编程初级实践
- Spark SQL基本操作
将下列JSON格式数据复制到Linux系统中，并保存命名为employee.json。
| { "id":1 , "name":" Ella" , "age":36 } { "id":2, "name":"Bob","age":29 } { "id":3 , "name":"Jack","age":29 } { "id":4 , "name":"Jim","age":28 } { "id":4 , "name":"Jim","age":28 } { "id":5 , "name":"Damon" } { "id":5 , "name":"Damon" } |
为employee.json创建DataFrame，并写出Scala语句完成下列操作：
- 查询所有数据；
- 查询所有数据，并去除重复的数据；
- 查询所有数据，打印时去除id字段；
- 筛选出age>30的记录；
- 将数据按age分组；
- 将数据按name升序排列；
- 取出前3行数据；
- 查询所有记录的name列，并为其取别名为username；
- 查询年龄age的平均值；
- 查询年龄age的最小值。
- 编程实现将RDD转换为DataFrame
源文件内容如下（包含id,name,age）：
| 1,Ella,36 2,Bob,29 3,Jack,29 |
请将数据复制保存到Linux系统中，命名为employee.txt，实现从RDD转换得到DataFrame，并按“id:1,name:Ella,age:36”的格式打印出DataFrame的所有数据。请写出程序代码。
- 编程实现利用DataFrame读写MySQL的数据
（1）在MySQL数据库中新建数据库sparktest，再创建表employee，包含如表6-2所示的两行数据。
表6-2 employee表原有数据
| id | name | gender | Age |
| 1 | Alice | F | 22 |
| 2 | John | M | 25 |
（2）配置Spark通过JDBC连接数据库MySQL，编程实现利用DataFrame插入如表6-3所示的两行数据到MySQL中，最后打印出age的最大值和age的总和。
表6-3 employee表新增数据
| id | name | gender | age |
| 3 | Mary | F | 26 |
| 4 | Tom | M | 23 |
实验一 ：Spark SQL基本操作
1）
// 导入必要的库
import org.apache.spark.sql.SparkSession
// 创建SparkSession
val spark = SparkSession.builder()
.appName("Spark SQL Basic Operations")
.getOrCreate()
// 读取JSON文件创建DataFrame
val df = spark.read.json("file:///home/hadoop/employee.json")
// (1) 查询所有数据
df.show()
（2）查询所有数据，并去除重复的数据
df.distinct().show()
（3）
查询所有数据，打印时去除id字段
df.drop("id").show()
（4）
筛选出age>30的记录
df.filter("age > 30").show()
（5）
将数据按age分组
df.groupBy("age").count().show()
（6）
将数据按name升序排列
df.orderBy("name").show()
（7）
取出前3行数据
df.limit(3).show()
（8）
查询所有记录的name列，并为其取别名为username
df.select($"name".alias("username")).show()
（9）
查询年龄age的平均值
df.selectExpr("avg(age)").show()
（10）
查询年龄age的最小值
df.selectExpr("min(age)").show()实验二 ：编程实现将RDD转换为DataFrame
编程代码：
import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.types._
object RDDToDataFrameExample {
def main(args: Array[String]): Unit = {
// 创建SparkSession
val spark = SparkSession.builder()
.appName("RDD to DataFrame Example")
.master("local[*]") // 使用本地模式，如果连接到集群请更改这里
.getOrCreate()
import spark.implicits._
// 指定employee.txt文件的位置
val inputFilePath = "file:///home/hadoop/employee.txt"
// 从文本文件读取数据创建RDD
val rdd = spark.sparkContext.textFile(inputFilePath)
// 定义DataFrame的schema
val schema = StructType(Array(
StructField("id", IntegerType, nullable = false),
StructField("name", StringType, nullable = false),
StructField("age", IntegerType, nullable = false)
))
// 将RDD转换为DataFrame
val dataFrame = spark.createDataFrame(rdd.map { line =>
val parts = line.split(",")
Row(parts(0).toInt, parts(1), parts(2).toInt)
}, schema)
// 显示DataFrame内容
dataFrame.show(false)
// 按照指定格式打印所有数据
dataFrame.collect().foreach { row =>
println(s"id:${row.getAs[Int]("id")},name:${row.getAs[String]("name")},age:${row.getAs[Int]("age")}")
}
// 停止SparkSession
spark.stop()
}
}命令
/usr/local/spark-3.5.1/bin/spark-submit --class "RDDToDataFrameExample" ./target/scala-2.12/simple-project_2.12-1.9.0.jar具体操作参考博客
如何安装sbt（sbt在ubuntu上的安装与配置）（有详细安装网站和图解）-CSDN博客
实验三：编程实现利用DataFrame读写MySQL的数据
mysql代码
CREATE DATABASE sparktest;
USE sparktest;
CREATE TABLE employee (
id INT PRIMARY KEY,
name VARCHAR(50),
gender CHAR(1),
age INT
);
INSERT INTO employee (id, name, gender, age) VALUES (1, 'Alice', 'F', 22);
INSERT INTO employee (id, name, gender, age) VALUES (2, 'John', 'M', 25);如何安装msyql参考博客
在ubuntu上安装mysql(在线安装需要)-CSDN博客
如何安装mysl驱动程序jar包-CSDN博客
编程代码
import org.apache.spark.sql.{SparkSession, Row}
import java.util.Properties
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.max
import org.apache.spark.sql.functions.sum
object MySQLDataFrameExample {
def main(args: Array[String]): Unit = {
// 创建SparkSession
val spark = SparkSession.builder()
.appName("MySQL DataFrame Example")
.master("local[*]") // 使用本地模式，如果连接到集群请更改这里
.getOrCreate()
import spark.implicits._
// 配置MySQL JDBC连接
val jdbcProperties = new Properties()
jdbcProperties.setProperty("user", "root")
jdbcProperties.setProperty("password", "mysql")
jdbcProperties.setProperty("driver", "com.mysql.cj.jdbc.Driver")
// 定义MySQL的JDBC连接URL
val jdbcUrl = "jdbc:mysql://localhost:3306/sparktest"
// 创建DataFrame以插入数据
val newEmployeeData = Seq(
(3, "Mary", "F", 26),
(4, "Tom", "M", 23)
).toDF("id", "name", "gender", "age")
// 将DataFrame数据插入到MySQL的employee表中
newEmployeeData.write
.mode("append") // 使用append模式来添加数据，而不是覆盖
.jdbc(jdbcUrl, "employee", jdbcProperties)
// 从MySQL读取employee表的数据
val employeeDF = spark.read
.jdbc(jdbcUrl, "employee", jdbcProperties)
// 打印age的最大值
val maxAge = employeeDF.agg(max("age")).collect()(0).getAs[Int](0)
println(s"Max age: $maxAge")
// 打印age的总和
val sumAge = employeeDF.agg(sum("age")).collect()(0).getAs[Long](0)
println(s"Sum of ages: $sumAge")
// 停止SparkSession
spark.stop()
}
}
编程详细步骤参考
如何安装sbt（sbt在ubuntu上的安装与配置）（有详细安装网站和图解）-CSDN博客
运行命令
/usr/local/spark-3.5.1/bin/spark-submit --jars /home/hadoop/mysql-connector-j-8.3.0/mysql-connector-j-8.3.0.jar --class "MySQLDataFrameExample" ./target/scala-2.12/simple-project_2.12-1.9.0.jar产生错误
主要问题都在实验三中，因为实验三中涉及到一个mysql数据库连接
命令更新为
/usr/local/spark-3.5.1/bin/spark-submit --jars /home/hadoop/mysql-connector-j-8.3.0/mysql-connector-j-8.3.0.jar --class "MySQLDataFrameExample" ./target/scala-2.12/simple-project_2.12-1.9.0.jar
加了一个mysl驱动的jar的引用
如何安装mysql驱动参考博客
如何安装mysl驱动程序jar包-CSDN博客
打包失败
这个问题是代码错误
代码未引入一些包
加上下面这些就可以了
import org.apache.spark.sql.{SparkSession, Row}
import java.util.Properties
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.max
import org.apache.spark.sql.functions.sum
运行失败
未引入mysl驱动程序
要下载mysql驱动
采用命令引入
/usr/local/spark-3.5.1/bin/spark-submit --jars /home/hadoop/mysql-connector-j-8.3.0/mysql-connector-j-8.3.0.jar --class "MySQLDataFrameExample" ./target/scala-2.12/simple-project_2.12-1.9.0.jar参考链接
如何安装sbt（sbt在ubuntu上的安装与配置）（有详细安装网站和图解）-CSDN博客
在ubuntu上安装mysql(在线安装需要)-CSDN博客
在ubuntu上安装mysql(在线安装需要)-CSDN博客
