头歌-Spark SQL 多数据源操作（Scala）

2025-11-26 19:35:07 分类：scala 热度：16 评论： 0

第1关：加载与保存操作

编程要求
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，完善程序。读取本地文件 file:///data/bigfiles/demo.json，根据年龄字段 age 设置降序，输出结果。

demo.json 文件内容如下所示：

{"name": "zhangsan", "age": 20, "sex": "m"},
{"name": "lisi", "age": 21, "sex": "m"},
{"name": "tiantian", "age": 22, "sex": "f"},
{"name": "lihua", "age": 23, "sex": "f"},
{"name": "zhaoliu", "age": 24, "sex": "m"},
{"name": "liguanqing", "age": 25, "sex": "f"},
{"name": "zhangqi", "age": 26, "sex": "m"},
{"name": "zhaoai", "age": 27, "sex": "m"},
{"name": "wangjiu", "age": 28, "sex": "f"}
开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object First_Question {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("First_Question")
      .master("local[*]")
      .getOrCreate()
    /******************* Begin *******************/
    val df: DataFrame = spark.read.json("file:///data/bigfiles/demo.json")
    val sortedDf = df.orderBy(df("age").desc)
    sortedDf.show()
    /******************* End *******************/
    spark.stop()
  }
}

第2关：Parquet 格式文件

编程要求
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，根据下列要求，完善程序。

读取本地文件 file:///data/bigfiles/demo.json，使用 Parquet 完成分区，列名为 student=1，保存到本地路径file:///result/下。

读取本地文件 file:///data/bigfiles/demo2.json，使用 Parquet 完成分区，列名为 student=2，保存到本地路径file:///result/下。

demo.json 文件内容如下所示：

{"name": "zhangsan", "age": 20, "sex": "m"},
{"name": "lisi", "age": 21, "sex": "m"},
{"name": "tiantian", "age": 22, "sex": "f"},
{"name": "lihua", "age": 23, "sex": "f"},
{"name": "zhaoliu", "age": 24, "sex": "m"},
{"name": "liguanqing", "age": 25, "sex": "f"},
{"name": "zhangqi", "age": 26, "sex": "m"},
{"name": "zhaoai", "age": 27, "sex": "m"},
{"name": "wangjiu", "age": 28, "sex": "f"}
demo2.json 文件内容如下所示：

{"name": "hongkong", "age": 20, "sex": "m"},
{"name": "kulu", "age": 21, "sex": "m"},
{"name": "huxiaotian", "age": 22, "sex": "f"},
{"name": "yueming", "age": 23, "sex": "f"},
{"name": "wangsan", "age": 24, "sex": "m"},
{"name": "zhaojiu", "age": 25, "sex": "f"},
{"name": "wangqiqi", "age": 26, "sex": "m"},
{"name": "wangxiantian", "age": 27, "sex": "m"},
{"name": "zhaoba", "age": 28, "sex": "f"}
开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object Second_Question {
  def main(args: Array[String]): Unit = {
    
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Second_Question")
      .master("local[*]")
      .getOrCreate()
 
    /******************* Begin *******************/  
    spark.read.json("file:///data/bigfiles/demo.json").write.parquet("file:///result/student=1") spark.read.json("file:///data/bigfiles/demo2.json").write.parquet("file:///result/student=2")
    /******************* End *******************/
    spark.stop()
 
  }
}

第3关：ORC 格式文件

编程要求
根据下列要求，完善程序。

创建 Orc 格式的 Hive 数据表 student，添加字段id（int），name（string）,age(int)，class（string）。

按顺序插入如下数据：

1001,"王刚",19,"大数据一班"
1002,"李虹",18,"大数据一班"
1003,"张子萱",20,"大数据一班"
1004,"赵云",18,"大数据一班"
1005,"李晓玲",19,"大数据一班"
1006,"张惠",18,"大数据二班"
1007,"秦散",19,"大数据二班"
1008,"王丽",18,"大数据二班"
1009,"田忌",20,"大数据二班"
1010,"张花",18,"大数据二班"
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，编写 spark sql 程序，读取创建的 student 表并按字段 id 升序输出。
开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object Third_Question {
  def main(args: Array[String]): Unit = {

    // 创建 SparkSession，并启用 Hive 支持
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Third_Question")
      .master("local[*]")
      .enableHiveSupport()  // 启用 Hive 支持
      .getOrCreate()

    /******************* Begin *******************/

    // 创建 Hive 数据表 student
    spark.sql(
      """
        |CREATE TABLE IF NOT EXISTS student (
        |  id INT,
        |  name STRING,
        |  age INT,
        |  class STRING
        |)
        |STORED AS ORC
      """.stripMargin)

    // 插入数据到 student 表
    spark.sql(
      """
        |INSERT INTO student VALUES
        |(1001, '王刚', 19, '大数据一班'),
        |(1002, '李虹', 18, '大数据一班'),
        |(1003, '张子萱', 20, '大数据一班'),
        |(1004, '赵云', 18, '大数据一班'),
        |(1005, '李晓玲', 19, '大数据一班'),
        |(1006, '张惠', 18, '大数据二班'),
        |(1007, '秦散', 19, '大数据二班'),
        |(1008, '王丽', 18, '大数据二班'),
        |(1009, '田忌', 20, '大数据二班'),
        |(1010, '张花', 18, '大数据二班')
      """.stripMargin)
    // 查询并按 id 字段升序输出 student 表数据
    val studentDF: DataFrame = spark.sql("SELECT * FROM student ORDER BY id ASC")
    studentDF.show()
    /******************* End *******************/
    spark.stop()
  }
}

第4关：JSON 格式文件

编程要求
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，完善程序。读取本地文件 file:///data/bigfiles/test.json，不改变原数据排列顺序进行输出。

test.json 文件内容如下所示：

{"id":1001,"name":"王刚","age":19,"class":"大数据一班"},
{"id":1002,"name":"李虹","age":18,"class":"大数据一班"},
{"id":1003,"name":"张子萱","age":20,"class":"大数据一班"},
{"id":1004,"name":"赵云","age":18,"class":"大数据一班"},
{"id":1005,"name":"李晓玲","age":19,"class":"大数据一班"},
{"id":1006,"name":"张惠","age":18,"class":"大数据二班"},
{"id":1007,"name":"秦散","age":19,"class":"大数据二班"},
{"id":1008,"name":"王丽","age":18,"class":"大数据二班"},
{"id":1009,"name":"田忌","age":20,"class":"大数据二班"},
{"id":1010,"name":"张花","age":18,"class":"大数据二班"}
开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object Forth_Question {
  def main(args: Array[String]): Unit = {
    
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Forth_Question")
      .master("local[*]")
      .getOrCreate()
    /******************* Begin *******************/  
    // 读取 JSON 文件
    val df = spark.read.json("file:///data/bigfiles/test.json")
    df.select("id", "name", "age", "class").show()

    /******************* End *******************/

    spark.stop()

  }
}

第5关：JDBC 操作数据库

编程要求
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，完善程序。读取本地 csv 文件 file:///data/bigfiles/job58_data.csv（有表头），将加载的数据以覆盖的方式保存到本地 Mysql 数据库的 work.job_data 表中，数据库连接信息如下：

账号：root

密码：123123

端口：3306

注意设置 useSSL=false。

开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Fifth_Question {
  def main(args: Array[String]): Unit = {
    
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Fifth_Question")
      .master("local[*]")
      .getOrCreate()
    /******************* Begin *******************/  
    // 1. 读取本地 CSV 文件
    val df: DataFrame = spark.read
      .option("header", "true") 
      .csv("file:///data/bigfiles/job58_data.csv")
    // 2. MySQL 连接信息
    val jdbcUrl = "jdbc:mysql://localhost:3306/work?useSSL=false"
    val dbProperties = new java.util.Properties()
    dbProperties.setProperty("user", "root")
    dbProperties.setProperty("password", "123123")
  
    // 3. 将数据写入 MySQL 数据库的 work.job_data 表中，覆盖原有数据
    df.write
      .mode(SaveMode.Overwrite) 
      .jdbc(jdbcUrl, "job_data", dbProperties)
    /******************* End *******************/

    spark.stop()

  }
}

第6关：Hive 表操作

编程要求
打开右侧代码文件窗口，在 Begin 至 End 区域补充代码，根据下列要求，完善程序。

在 Hive 中创建数据表 employee，添加字段eid（string），ename（string）,age(int)，part（string）。

插入如下数据：

"A568952","王晓",25,"财务部"
"B256412","张天",28,"人事部"
"C125754","田笑笑",23,"销售部"
"D265412","赵云",24,"研发部"
"F256875","李姿姿",26,"后勤部"
编写 spark sql 程序，直接采用 Spark on Hive 的方式读取创建的 employee 表并按字段 eid 升序输出。
开始任务前，注意先启动 Hadoop 与 Hive 环境：start-all.sh、nohup hive --service metastore &

import org.apache.spark.sql.{DataFrame, SparkSession}

object Sixth_Question {
  def main(args: Array[String]): Unit = {
    /******************* Begin *******************/  
    try {
      // 创建SparkSession，启用Hive支持
      val spark = SparkSession.builder()
        .appName("ReadHiveTable")
        .enableHiveSupport()
        .getOrCreate()
        
      import spark.implicits._
      import spark.sql

      // 如果表不存在则创建
      if (!spark.catalog.tableExists("employee")) {
        sql("""
          CREATE TABLE employee (
            eid STRING, 
            ename STRING, 
            age INT, 
            part STRING
          )
        """)
        
        // 准备数据
        val employeeData = Seq(
          ("A568952","王晓",25,"财务部"),
          ("B256412","张天",28,"人事部"),
          ("C125754","田笑笑",23,"销售部"),
          ("D265412","赵云",24,"研发部"),
          ("F256875","李姿姿",26,"后勤部")
        )
        
        // 转换为DataFrame并写入Hive表
        employeeData.toDF("eid", "ename", "age", "part")
          .write.mode("append").saveAsTable("employee")
      }
      // 读取Hive表并按eid升序排序
      val employeeDF: DataFrame = sql("SELECT * FROM employee ORDER BY eid ASC")
      
      // 显示结果
      println("")
      employeeDF.show()
      
      // 停止SparkSession
      spark.stop()
    } catch {
      case e: Exception =>
        println(s"程序执行出错: ${e.getMessage}")
        e.printStackTrace()
    }
    /******************* End *******************/

  }
}

转载请说明出处内容投诉
CSS教程网 » 头歌-Spark SQL 多数据源操作（Scala）
心舞飞扬

心舞飞扬

分享到：

发表评论

欢迎 访客 发表评论

一个令你着迷的主题！

查看演示 官网购买