更多 Spark DataFrame 概念

PySpark 入門

Benjamin Schmidt

Data Engineer

從多種資料來源建立 DataFrame

CSV 檔：常見的結構化、分隔資料
JSON 檔：半結構化、階層式格式
Parquet 檔：為儲存與查詢最佳化，常用於資料工程

範例：
```
spark.read.csv("path/to/file.csv")
```
範例：
```
spark.read.json("path/to/file.json")
```

範例：

spark.read.parquet("path/to/file.parquet")

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.pandas/api/pyspark.pandas.read_csv

自動推斷與手動定義綱要（Schema）

Spark 可用 inferSchema=True 從資料推斷綱要
也可手動定義綱要以更精準控制——適合固定結構

大規模綱要

PySpark DataFrame 的資料型別（DataTypes）

IntegerType：整數
- 例如：1、3478、-1890456
LongType：較大的整數
- 例如：8 位元組帶號數字，922334775806
FloatType 與 DoubleType：浮點數（小數）
- 例如：3.14159
StringType：文字（字串）
- 例如："This is an example of a string."
…

PySpark DataFrame 的資料型別語法

# 匯入所需型別類別
from pyspark.sql.types import (StructType,
                            StructField, IntegerType,
                            StringType, ArrayType)

# 建立綱要（schema）
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("scores", ArrayType(IntegerType()), True)
])

# 套用綱要
df = spark.createDataFrame(data, schema=schema)

DataFrame 操作：選取與篩選

用 .select() 選取特定欄位
用 .filter() 或 .where() 依條件篩選列
用 .sort() 依多欄位排序

# 只選取並顯示 name 與 age 欄位
df.select("name", "age").show()

# 篩選 age > 30
df.filter(df["age"] > 30).show()

# 使用 where 篩選特定值
df.where(df["age"] == 30).show()

# 使用 sort 依年齡排序
df.sort("age", ascending=False).show()

排序與移除遺漏值

使用 .sort() 或 .orderBy() 排序
用 na.drop() 移除含 null 的列

# 以 age 欄位排序
df.sort("age", ascending=False).show()

# 移除遺漏值
df.na.drop().show()

小抄

spark.read_json()：從 JSON 載入資料
spark.read.schema()：明確定義綱要
.na.drop()：刪除含遺漏值的列
.select()、.filter()、.sort()、.orderBy()：基本資料操作

一起來練習吧！

PySpark 入門