一、前置环境的设置

1.启动hadoop集群

start-dfs.sh
start-yarn.sh

2.查看hdfs是否开启安全模式

如果安全 模式打开，则需要输入命令关闭

hdfs dfsadmin -safemode leave

二、房屋销售数据处理

1. 数据的读入

（1）关闭日志显示功能

 Logger.getLogger("org").setLevel(Level.OFF)

（2）创建SparkSession对象

 val spark = SparkSession
      .builder()
      //应用程序提交的模式
      .master("local")
      //给应用程序起名
      .appName("house")
      .getOrCreate()

（3）导入隐式包

  import spark.implicits._

（4）csv.数据文件的读取

 val houseDf = spark
      .read
      //读取第一行数据作为表头
      .option("header","true")
      //对字段的数据类型做自动推断
      .option("inferSchema","true")
      .csv("hdfs://master:8020/spark/data/house.csv")

（5）显示数据

houseDf.show()

（6）打印表结构

 houseDf.printSchema()

（7）停止SparkSession对象

 spark.stop()

(8)完整示例

 //1.日志显示功能的关闭
    Logger.getLogger("org").setLevel(Level.OFF)

    //2.创建SparkSession对象——sparksql的编程入口
    val spark = SparkSession
      .builder()
      //应用程序提交的模式
      .master("local")
      //给应用程序起名
      .appName("house")
      .getOrCreate()

    //3.导入隐式包
    import spark.implicits._

    //4.读取数据----DataFrame
    val house = spark
      .read
      //将第一行读入为表头信息
      .option("header","true")
      .option("inferSchema","true")
      .csv("hdfs://master:8020/spark/data/house.csv")

    //5.查看数据及表结构
    house.show()
    house.printSchema()
    //最后：sparksql编程入口关闭
    spark.stop()

2.统计每个季度的房屋销量——房屋的数量

(1)相关函数的知识点

A.withColumn函数

withColumn 是 SparkSQL 中用于添加或替换 DataFrame 列的核心函数，通过表达式或用户定义函数（UDF）生成新列。以下内容涵盖其用法、注意事项及示例。

基本语法

def withColumn(colName: String, col: Column): DataFrame

• colName: 新列或待替换列的名称。
• col: 生成列值的表达式（如 col("existing") + 1 或 UDF）。

典型应用场景

添加新列
通过现有列计算新列：

df.withColumn("age_plus_10",$"age" + 10)

替换现有列
修改列值或类型：

df.withColumn("age", $"age".cast(StringType))

条件赋值
结合 when 和 otherwise 实现条件逻辑：

df.withColumn("age_group", 
    when($"age" < 18, "minor").otherwise("adult"))

性能优化建议

• 避免链式调用：多次调用 withColumn 会生成多个中间 DataFrame，建议合并操作或使用 select。
• 优先使用原生函数：内置函数（如 concat、substring）比 UDF 效率更高。
• 广播小数据集：在 UDF 中引用小数据集时，使用广播变量减少数据传输。

常见错误与解决

列名冲突
若新列名与现有列名重复，原列会被覆盖。需确保命名唯一性。

类型不匹配
表达式返回值类型需与目标列兼容。例如，以下代码会报错：

df.withColumn("age", concat($"age", " years")) // age 原为整数类型

解决方案是显式转换类型：

df.withColumn("age", concat($"age".cast(StringType), " years"))

UDF 未注册
使用 UDF 前需在 SparkSession 中注册：

val toUpper = udf((s: String) => s.toUpperCase)
spark.udf.register("toUpper", toUpper)
df.withColumn("name_upper", expr("toUpper(name)"))

B. to_date函数的基本用法

在SparkSQL中，to_date函数用于将字符串或时间戳转换为日期类型（DateType）。其基本语法如下：

to_date(date_str,format)

• date_str：需要转换的字符串或时间戳列。
• format（可选）：指定输入字符串的格式。默认为yyyy-MM-dd或yyyy-MM-dd HH:mm:ss。

格式控制字符串的书写规则

		    y——年
            M——月
            d——日
            H——时
            m——分
            s——秒

转换自定义格式字符串

若字符串格式非标准，需通过format参数明确指定：

import org.apache.spark.sql.functions.to_date
			 val time = spark
      				.createDataFrame(List(("tom","2020-05-05 12:25:34"),("bob","2021-06-28 										12:25:34"),("jerry","2022-06-18 12:25:34")))
      .toDF("name","birthday")
      time
      .withColumn("birthday",to_date($"birthday","yyyy-MM-dd HH:mm:ss"))

常用格式符号

符号	含义	示例
yyyy	四位年份	2023
MM	两位月份	10
dd	两位日期	01
HH	24小时制小时	14

性能注意事项

• 对大规模数据使用固定格式比自动解析更高效。
• 建议在读取数据后尽早执行类型转换，减少后续计算的类型推断开销。

C.时间值处理函数

以下是 SparkSQL 中常用的时间处理函数及其功能说明，适用于 Scala 编程语言：

---

基础日期提取函数

year
提取日期或时间戳中的年份部分，返回整数。
示例：year("2023-05-20") → 返回 2023。

month
提取日期或时间戳中的月份部分（1-12）。
示例：month("2023-05-20") → 返回 5。

dayofyear
返回日期在一年中的天数（1-366）。
示例：dayofyear("2023-02-01") → 返回 32（第32天）。

dayofmonth
提取日期中的“日”部分（1-31）。
示例：dayofmonth("2023-05-20") → 返回 20。

---

周和季度相关函数

weekofyear
返回日期所在的周数（1-53）。
示例：weekofyear("2023-01-01") → 返回 1。

quarter
返回日期所在的季度（1-4）。
示例：quarter("2023-05-20") → 返回 2（第二季度）。

---

时间戳处理函数

hour
提取时间戳中的小时部分（0-23）。
示例：hour("2023-05-20 14:30:00") → 返回 14。

minute
提取时间戳中的分钟部分（0-59）。
示例：minute("2023-05-20 14:30:00") → 返回 30。

second
提取时间戳中的秒部分（0-59）。
示例：second("2023-05-20 14:30:45") → 返回 45。

---

日期计算与转换

date_add
在日期上增加指定的天数。
示例：date_add("2023-05-20", 3) → 返回 2023-05-23。

date_sub
在日期上减少指定的天数。
示例：date_sub("2023-05-20", 3) → 返回 2023-05-17。

datediff
计算两个日期之间的天数差。
示例：datediff("2023-05-25", "2023-05-20") → 返回 5。

---

字符串与日期转换

to_date
将字符串转换为日期格式。
示例：to_date("20230520", "yyyyMMdd") → 返回 2023-05-20。

date_format
将日期按指定格式转换为字符串。
示例：date_format("2023-05-20", "MM/dd/yyyy") → 返回 "05/20/2023"。

---

高级函数

last_day
返回日期所在月份的最后一天。
示例：last_day("2023-02-15") → 返回 2023-02-28。

next_day
返回指定日期后的下一个星期几的日期。
示例：next_day("2023-05-20", "Mon") → 返回下周一日期。

---

实际应用示例（Scala代码）

import org.apache.spark.sql.functions._

// 创建DataFrame
val df = List(("2023-05-20 14:30:45"), ("2022-12-31")).toDF("timestamp")

// 使用时间函数处理
val result = df.withColumn("year", year(to_timestamp($"timestamp")))
              .withColumn("month", month(to_timestamp($"timestamp")))
              .withColumn("dayofyear", dayofyear(to_date($"timestamp")))

result.show()

输出结果：

+-------------------+----+-----+---------+
|          timestamp|year|month|dayofyear|
+-------------------+----+-----+---------+
|2023-05-20 14:30:45|2023|    5|      140|
|       2022-12-31  |2022|   12|      365|
+-------------------+----+-----+---------+

（2）任务实现

 val cleanDf = house
    //对字段做数据类型的转换——withColumn   integer--->string
      .withColumn("sale_data",$"sale_data".cast(StringType))
      //string--->date  20200302   yyyyMMdd
      .withColumn("sale_data",to_date($"sale_data","yyyyMMdd"))
   cleanDf
      .withColumn("quarter",quarter($"sale_data"))
      .groupBy("quarter")
      .count()
      .show()