Spark Dataset：用类型系统消灭一次"字段名拼错"的线上事故

val region = df.getString(0)   // <-- 取第一列

case class Order(id: Long, region: String, amount: Double)
val orders: Dataset[Order] = df.as[Order]

// 拿 region 字段
val region = df.getString(0)            // <-- 实际是拿"id"列，因为字段顺序变了

val orders: Dataset[Order] = df.as[Order]
val region = orders.head.region // <-- 直接拿 region 字段，编译器检查类型

val orders: Dataset[Order] = df.as[Order]
val x = orders.head.regin   // <-- 编译错误：value regin is not a member of Order

case class Person(name: String, age: Int)
val ds: Dataset[Person] = Seq(Person("Alice", 30), Person("Bob", 25)).toDS()

ds.show()
/*
+-----+---+
| name|age|
+-----+---+
|Alice| 30|
|  Bob| 25|
+-----+---+
*/

// 编译器会查找隐式的 Encoder[Person]
//   对于 case class，默认是 ExpressionEncoder
//   它把 Person 编码成 Spark 内部的 InternalRow
//   取数据时再从 InternalRow 解码回 Person

val ds: Dataset[Int] = Seq(1, 2, 3).toDS() // <-- Dataset[Int]，基本类型也有 Encoder
val ds2: Dataset[(String, Int)] = Seq("a" -> 1, "b" -> 2).toDS()  // <-- Tuple 也有

import org.apache.spark.sql.Encoder
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder

case class Money(amount: BigDecimal, currency: String)

// 用 ExpressionEncoder 声明
implicit val moneyEncoder: Encoder[Money] = ExpressionEncoder[Money]

// 现在 Money 类型能用了
val ds: Dataset[Money] = Seq(Money(BigDecimal(100), "USD")).toDS()

// ❌ 错的写法
import java.sql.Date
case class Order(id: Long, orderDate: Date)

// 读出来怎么是 1970-01-01？
val ds: Dataset[Order] = df.as[Order]    // <-- df 里是 LocalDate，类型不匹配

case class Address(city: String, zip: String)
case class Person(name: String, age: Int, addresses: Seq[Address])

val people: Dataset[Person] = Seq(
  Person("Alice", 30, Seq(Address("Beijing", "100000"), Address("Shanghai", "200000")))
).toDS()

people.show(false)
/*
+-----+---+-----------------------------------------+
|name |age|addresses                                |
+-----+---+-----------------------------------------+
|Alice| 30|[[Beijing, 100000], [Shanghai, 200000]] |
+-----+---+-----------------------------------------+
*/

// 访问嵌套字段
people.select($"addresses".getField("city")) // <-- 用 .getField 取 struct 子字段

// 展开数组
people.select(explode($"addresses").as("addr")) // <-- explode 把 array 展开成多行

case class Person(name: String, age: Option[Int])  // <-- age 可能是 null

val p = Person("Alice", Some(30))
val p2 = Person("Bob", None)

// 取值时
p.age.getOrElse(0)        // <-- 30
p2.age.getOrElse(0)       // <-- 0

// 用 Option 写
case class Person(name: String, age: Option[Int])
def getAge(p: Person): Int = p.age match {
  case Some(age) => age
  case None      => 0
}

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

// 处理 null 的几种函数
df.select(coalesce($"col", "default"))                     // <-- 返回第一个非 null
df.select(nvl($"col", "default"))                          // <-- null 替换为默认值（coalesce 的两参数版本）
df.select(nvl2($"col", "has_value", "is_null"))            // <-- 三元：null vs not-null

// 过滤 null
df.filter($"col".isNull)        // <-- 取 null 的行
df.filter($"col".isNotNull)     // <-- 取非 null 的行

// ❌ NaN == NaN 是 false
df.filter($"score" === Double.NaN)   // <-- 一个都过滤不出来

// ✅ 用 isNaN
df.filter($"score".isNaN)            // <-- 正确

// DataFrame 转 Dataset
case class Person(name: String, age: Int)
val df = spark.read.parquet("path")
val ds: Dataset[Person] = df.as[Person]            // <-- 转成 Dataset[Person]

// Dataset 转 DataFrame
val df2: DataFrame = ds.toDF()                      // <-- 回到 DataFrame

// 隐式转换
import spark.implicits._
val ds2: Dataset[Person] = df.as[Person]           // <-- 跟前面等价

// DataFrame = Dataset[Row]——类型是 Row，无 schema
type DataFrame = Dataset[Row]

// Dataset[T]——类型是 T，有 schema
val ds: Dataset[Person]