Spark DataFrame 實戰:讀取、轉換、寫出與 Spark SQL

上一篇把 Spark 的概念講完了:Driver / Executor、分區、惰性求值、shuffle。這篇動手用 DataFrame 實際轉一份資料 —— 讀進來、清洗、彙總、寫出去,並看看 Spark SQL 怎麼和 DataFrame API 殊途同歸。

為什麼幾乎都用 DataFrame,而不是 RDD

上一篇提過 RDD 是最底層的抽象。但日常工作九成用 DataFrame,因為它有兩個 RDD 沒有的東西:

• Schema(欄位與型別):像一張有結構的表,讀起來像在寫 SQL。
• Catalyst 最佳化器:你寫的是「要什麼」,Catalyst 幫你決定「怎麼做最快」(調換順序、下推 filter…)。RDD 是你叫它怎麼做就怎麼做,沒人幫你最佳化。

換句話說:DataFrame 是宣告式、被最佳化的;RDD 是命令式、要自己顧效能。 除非你需要極細粒度的控制,否則 DataFrame 又快又好讀。

讀取:給它 schema 比較好

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("orders").getOrCreate()

# parquet 自帶 schema 與型別,比 csv inferSchema 快又準
df = spark.read.parquet("orders.parquet")
df.printSchema()   # 看欄位與型別

CSV 也能讀(spark.read.csv(..., header=True, inferSchema=True)),但 inferSchema 要多掃一次資料、型別還可能猜錯;正式的 pipeline 我會用 parquet 或明確給 schema。

常用轉換:像操作一張表

monthly = (
    df.filter(F.col("status") == "paid")                         # 篩選(narrow)
      .withColumn("month", F.date_format("order_date", "yyyy-MM"))  # 新增欄位
      .groupBy("month")                                          # 分組(wide,要 shuffle)
      .agg(
          F.sum("amount").alias("revenue"),
          F.countDistinct("customer_id").alias("buyers"),
      )
      .orderBy("month")
)
monthly.show()

select / withColumn / filter 這些是 narrow:每個分區自己做,不用跨節點搬資料,幾乎免費。groupBy / join 是 wide:要把同一組的資料拉到一起,就得 shuffle —— 這是 上一篇說的「最貴的操作」。

Spark SQL:同一份資料,換 SQL 寫

把 DataFrame 註冊成一張臨時表,就能直接寫 SQL —— 結果跟上面的 DataFrame 寫法完全等價:

df.createOrReplaceTempView("orders")

monthly = spark.sql("""
    select date_format(order_date, 'yyyy-MM') as month,
           sum(amount)                        as revenue,
           count(distinct customer_id)        as buyers
    from orders
    where status = 'paid'
    group by 1
    order by 1
""")

為什麼兩種寫法等價?因為它們都被 Catalyst 編成同一個實體執行計畫:

寫出:用 partitionBy 讓重跑變冪等

monthly.write \
    .mode("overwrite") \
    .partitionBy("month") \
    .parquet("out/monthly")

partitionBy("month") 會把輸出依月份切成不同資料夾;搭配 mode("overwrite"),重跑某個月就只覆寫那個月的分區。這正好呼應 Airflow 排程那篇講的冪等:同一段資料重算,結果蓋掉舊的、不會變兩倍。Spark 負責算,Airflow 負責排程觸發,兩邊都守住冪等,整條 pipeline 才補得了、重跑得起。

反思

DataFrame 和 SQL 不要二選一,挑可讀性

我不太理解「一定要用 DataFrame API」或「一定要寫 SQL」的教條。它們既然編成同一個計畫、效能一樣,那選擇標準就只有一個:哪個讓這段邏輯更好讀、更好維護。複雜的多步轉換、要拆函式重用,我用 DataFrame API;一段直白的彙總查詢,SQL 反而更短更清楚 —— 而且團隊裡每個人都看得懂 SQL。能混用就混用,別自我設限。

效能直覺,八成還是回到 shuffle

寫了一堆 DataFrame 操作後,真正會拖慢的幾乎都是 wide 轉換。我的習慣是先問自己三件事:能不能先 filter 把資料變小再 groupBy/join?join 的另一邊是不是小表、可以 broadcast 掉 shuffle?輸出分區數是不是合理?把這幾個顧好,比起糾結 select 怎麼寫,效益大得多。這跟 上一篇的結論一致:少 shuffle 就是 Spark 效能的本體。

宣告式的好處,是把最佳化交給比你聰明的東西

DataFrame 最值得體會的一點,是「我描述要什麼,Catalyst 決定怎麼做」。這跟自己用 RDD 手刻每一步、自己顧順序,是兩種心態。多數時候,相信最佳化器、把心力放在「邏輯對不對、shuffle 多不多」,比逐行微調划算 —— 這也是現代資料工具的共同方向:把人從實作細節解放出來,專注在意圖。