第12课：PySpark DataFrame多种创建方式、数据结构与基础操作实战

文章目录

• • 一、课前导读
  • 二、学习目标
  • 三、核心理论知识点
  • 四、原理通俗讲解
  • • 4.1 DataFrame的创建：数据源的“通用转换器”
    • 4.2 DataFrame的数据结构：二维表 + 强类型元数据
    • 4.3 DataFrame的操作：构建逻辑计划树
  • 五、重点概念拆解
  • • 5.1 创建DataFrame的各种方式
    • • 5.1.1 从本地集合创建
      • 5.1.2 从RDD创建
      • 5.1.3 从外部文件创建
      • 5.1.4 从JDBC创建
      • 5.1.5 从Hive表创建
    • 5.2 DataFrame的重要属性
    • 5.3 常用基础操作
    • 5.4 列表达式详解
    • 5.5 内置函数
  • 六、易错点避坑
  • • 6.1 频繁使用`collect()`或`toLocalIterator()`
    • 6.2 混淆`filter`和`where`
    • 6.3 误用`selectExpr`
    • 6.4 忽视Schema定义导致的性能问题
    • 6.5 未处理CSV中的空值或坏记录
    • 6.6 对DataFrame进行逐行操作
  • 七、完整实战案例
  • 八、代码逐行解析
  • • 8.1 创建DataFrame部分
    • 8.2 合并DataFrame
    • 8.3 列操作
    • 8.4 过滤
    • 8.5 聚合与连接
    • 8.6 缺失值处理
  • 九、业务场景落地应用
  • • 9.1 场景一：数据湖中的批量数据读取
    • 9.2 场景二：多源数据融合
    • 9.3 场景三：数据质量检查
    • 9.4 场景四：生成报表并导出
  • 十、常见报错排查
  • • 10.1 `AnalysisException: cannot resolve 'xxx'`
    • 10.2 `TypeError: condition should be string or Column`
    • 10.3 `Py4JError` 在`collect`时
    • 10.4 读取CSV时数据全变为`null`
    • 10.5 `org.apache.spark.SparkException: Job aborted due to stage failure`
  • 十一、本节课知识点总结
  • • DataFrame创建方式一览
    • 常用操作速查
    • 性能最佳实践
  • 十二、课后思考作业
  • • 作业一：理论理解题
    • 作业二：代码实践题
    • 作业三：场景应用题
    • 作业四：拓展研究
• 🔗《20节课 PySpark 从入门到精通》系列课程导航

一、课前导读

在上一节课中，我们了解了Spark SQL的整体架构以及DataFrame相比RDD的巨大优势。你已经知道DataFrame是带Schema的分布式数据集，并且Catalyst优化器能让它跑得飞快。但你可能仍然有一个核心疑问：“我该如何在实际工作中创建DataFrame？创建之后，又该如何操作它？”

这是一个非常实际的问题。在企业大数据开发中，数据源五花八门：业务系统的MySQL表、埋点日志的JSON文件、数据仓库的Hive表、实时消息队列的流数据……如果不能熟练地从各种数据源创建DataFrame，后续的分析就无从谈起。同样，DataFrame提供了极其丰富的API——select、filter、groupBy、join、withColumn等，如果只掌握皮毛，写出的代码可能又长又难读，甚至性能低下。

本节课就是你的“DataFrame操练手册”。我们将系统讲解DataFrame的多种创建方式（从集合、RDD、文件、JDBC、Hive等），深入剖析DataFrame的内部数据结构（Row、Column、StructType），并通过大量实战案例带你掌握最常用的基础操作。学完这节课，你将能够流畅地将任何数据源转为DataFrame，并使用DSL风格进行数据清洗和转换。

二、学习目标

完成本节课的学习后，你将能够：

三、核心理论知识点

四、原理通俗讲解

4.1 DataFrame的创建：数据源的“通用转换器”

可以把SparkSession看作一个“数据转换工厂”，它提供了read接口，能够从各种数据源读取数据，统一转换为DataFrame。无论源文件是CSV、JSON、Parquet，还是关系型数据库，你都可以用几乎相同的代码完成读取。

例如：

df_csv = spark.read.csv("path/to/file.csv", header=True)
df_json = spark.read.json("path/to/file.json")
df_parquet = spark.read.parquet("path/to/file.parquet")
df_jdbc = spark.read.format("jdbc").option("url", "jdbc:mysql://…").option("dbtable", "table").load()

这种统一的API大大降低了学习成本。同时，write接口也对称地支持多种格式输出。

4.2 DataFrame的数据结构：二维表 + 强类型元数据

DataFrame在逻辑上就是一张数据库表：行（Row）和列（Column）。每个列都有明确的数据类型（String、Integer、Double、Array、Map、Struct等）。这种带Schema的结构使得Spark可以进行很多智能优化，比如只读取需要的列（列剪枝），只读取符合条件的分区（分区剪枝）等。

物理上，DataFrame底层是由RDD[Row] + Schema信息组成的，但Row对象并不存储列的类型信息——类型信息完全由StructType管理。这意味着在访问列数据时，你需要按照正确的类型提取，例如row.getInt(0)或row["age"]返回的是Python对象但带类型。

4.3 DataFrame的操作：构建逻辑计划树

当你写df.select("name").filter(df.age > 18)时，Spark并不是立即执行，而是构建一棵抽象语法树（AST）。例如上述代码会生成一个逻辑计划：Project(name) ← Filter(age > 18) ← DataFrame。Catalyst优化器会改写这棵树，比如将Filter下推到数据源（谓词下推），然后生成物理计划并执行。整个过程对用户透明，但你写的代码越简洁，优化器越容易发挥威力。

五、重点概念拆解

5.1 创建DataFrame的各种方式

5.1.1 从本地集合创建

# 方式1：自动推断类型
data = [(1, "Alice", 25), (2, "Bob", 30)]
df = spark.createDataFrame(data, schema=["id", "name", "age"])

# 方式2：显式指定Schema
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
schema = StructType([
StructField("id", IntegerType(), True),
StructField("name", StringType(), True),
StructField("age", IntegerType(), True)
])
df = spark.createDataFrame(data, schema)

5.1.2 从RDD创建

rdd = sc.parallelize([(1, "Alice", 25), (2, "Bob", 30)])
# 方法1：toDF
df = rdd.toDF(["id", "name", "age"])
# 方法2：createDataFrame
df = spark.createDataFrame(rdd, schema)

5.1.3 从外部文件创建

# CSV
df = spark.read.csv("/path/file.csv", header=True, inferSchema=True)
# JSON
df = spark.read.json("/path/*.json")
# Parquet（最推荐）
df = spark.read.parquet("/path/table.parquet")
# Text（每行一个字符串）
df = spark.read.text("/path/file.txt")

5.1.4 从JDBC创建

df = spark.read.format("jdbc") \\
.option("url", "jdbc:mysql://localhost:3306/db") \\
.option("dbtable", "table") \\
.option("user", "root") \\
.option("password", "pwd") \\
.load()

5.1.5 从Hive表创建

spark.sql("USE mydatabase")
df = spark.table("my_table")
# 或直接SQL
df = spark.sql("SELECT * FROM my_table")

5.2 DataFrame的重要属性

• df.columns：返回列名列表
• df.dtypes：返回(列名, 类型)列表
• df.schema：返回StructType对象
• df.isLocal：是否本地（测试用）
• df.rdd：底层的RDD

5.3 常用基础操作

5.4 列表达式详解

在DataFrame操作中，列可以用多种方式表示：

• 字符串："age"
• col函数：col("age")
• df["age"]
• df.age（不推荐，容易与DataFrame方法冲突）
• 字符串表达式：expr("age + 1")

其中col和expr最灵活，expr可以写复杂表达式，如expr("CASE WHEN age > 18 THEN 'adult' ELSE 'minor' END")。

5.5 内置函数

pyspark.sql.functions提供了丰富的内置函数，例如：

• 聚合：sum, avg, count, min, max, collect_list, collect_set
• 窗口：row_number, rank, lag, lead
• 集合：array_contains, size, explode
• 字符串：concat, split, regexp_extract, trim, upper, lower
• 日期：to_date, year, month, date_add
• 条件：when, otherwise, coalesce, isnull

使用内置函数比UDF快几个数量级，应优先使用。

六、易错点避坑

6.1 频繁使用collect()或toLocalIterator()

在大数据集上调用collect()会将所有数据拉取到Driver内存，极易OOM。仅在确认结果很小或为了调试时使用。

6.2 混淆filter和where

两者完全等价，where是filter的别名，可以根据习惯选择。

6.3 误用selectExpr

selectExpr接受SQL表达式字符串，如df.selectExpr("id", "age + 1 as age_plus_one")，比select加alias更简洁，但要注意SQL注入风险（一般内部使用没问题）。

6.4 忽视Schema定义导致的性能问题

如果从RDD创建DataFrame时不提供Schema，Spark会通过采样（sampleRatio参数控制）来推断类型。对于大数据集，采样开销大且可能推断不准。最佳实践是显式定义StructType。

6.5 未处理CSV中的空值或坏记录

CSV读取时，默认将无法解析的值转为null，但你可能需要更多控制：.option("mode", "DROPMALFORMED")或.option("mode", "FAILFAST")。

6.6 对DataFrame进行逐行操作

如使用df.foreach(lambda row: …)效率极低，因为会触发大量对象创建。应使用foreachPartition批量操作或直接使用DataFrame内置函数。

七、完整实战案例

本案例将演示从各种数据源创建DataFrame，并执行常见的数据清洗和分析操作，覆盖select、filter、groupBy、join、withColumn、dropDuplicates等，最后保存结果。

# ============== dataframe_basics_demo.py ==============
# 功能：PySpark DataFrame 多种创建方式与基础操作实战
# 涵盖：创建、结构查看、选择、过滤、添加列、聚合、连接、写入等

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType, ArrayType
from pyspark.sql.functions import col, expr, when, avg, count, sum as spark_sum, concat, lit, year, month
import random
import os

# ========== 1. 创建SparkSession ==========
spark = SparkSession.builder \\
.appName("DataFrameBasics") \\
.master("local[4]") \\
.config("spark.sql.shuffle.partitions", "4") \\
.getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("WARN")

print("=" * 80)
print("PySpark DataFrame 多种创建方式与基础操作实战")
print("=" * 80)

# ========== 2. 创建DataFrame的各种方式 ==========
print("\\n步骤1: 创建DataFrame")
print("-" * 60)

# 2.1 从列表创建（自动推断Schema）
data1 = [(1, "Alice", 100, "HR"), (2, "Bob", 200, "IT"), (3, "Cathy", 150, "IT")]
df1 = spark.createDataFrame(data1, schema=["id", "name", "score", "dept"])
print("方式1: 从列表+列名创建")
df1.show()

# 2.2 显式定义Schema（推荐，性能好且精确）
schema2 = StructType([
StructField("id", IntegerType(), False),
StructField("name", StringType(), True),
StructField("salary", DoubleType(), True),
StructField("dept", StringType(), True)
])
data2 = [(4, "David", 2500.0, "Finance"), (5, "Ella", 3000.0, "HR")]
df2 = spark.createDataFrame(data2, schema2)
print("方式2: 显式Schema")
df2.printSchema()
df2.show()

# 2.3 从RDD[Row]创建
from pyspark.sql import Row
rdd = sc.parallelize([Row(id=6, name="Frank", salary=2800.0, dept="IT"),
Row(id=7, name="Grace", salary=3100.0, dept="HR")])
df3 = spark.createDataFrame(rdd)
print("方式3: 从RDD[Row]创建")
df3.show()

# 2.4 从RDD[tuple] + toDF
rdd_tup = sc.parallelize([(8, "Henry", 2700.0, "Finance"), (9, "Ivy", 2900.0, "IT")])
df4 = rdd_tup.toDF(["id", "name", "salary", "dept"])
print("方式4: RDD.toDF")
df4.show()

# 2.5 从CSV文件（先准备一个测试CSV）
csv_path = "/tmp/employees.csv"
with open(csv_path, "w") as f:
f.write("id,name,salary,dept\\n")
f.write("10,John,3200,HR\\n11,Jane,2800,IT\\n12,Jack,4000,Finance\\n")
df_csv = spark.read.csv(csv_path, header=True, inferSchema=True)
print("方式5: 从CSV读取")
df_csv.show()

# 2.6 从JSON文件（生成简单JSON）
json_path = "/tmp/employees.json"
with open(json_path, "w") as f:
f.write('{"id":13,"name":"Kevin","salary":3600,"dept":"IT"}\\n')
f.write('{"id":14,"name":"Lisa","salary":4100,"dept":"HR"}\\n')
df_json = spark.read.json(json_path)
print("方式6: 从JSON读取")
df_json.show()

# 2.7 从Parquet（先写入再读，演示）
parquet_path = "/tmp/employees_parquet"
df1.write.mode("overwrite").parquet(parquet_path)
df_parquet = spark.read.parquet(parquet_path)
print("方式7: 从Parquet读取")
df_parquet.show(2)

# ========== 3. 合并多个DataFrame（union） ==========
print("\\n" + "=" * 80)
print("步骤2: 合并DataFrame (union)")
print("=" * 80)

# 注意：union要求两个DataFrame的列数、顺序、类型一致，不一致需先select调整
# 统一列顺序
df1_aligned = df1.select("id", "name", "score", "dept")
df2_aligned = df2.select("id", "name", "salary", "dept").withColumnRenamed("salary", "score")
df_combined = df1_aligned.union(df2_aligned)
print("合并后DataFrame:")
df_combined.show()

# ========== 4. 基础操作：查看Schema和数据 ==========
print("\\n" + "=" * 80)
print("步骤3: 查看数据结构")
print("=" * 80)
print("列名:", df_combined.columns)
print("数据类型:", df_combined.dtypes)
print("Schema详细:")
df_combined.printSchema()
print("前3行数据:")
df_combined.show(3, truncate=False)

# ========== 5. 选择列和表达式 ==========
print("\\n" + "=" * 80)
print("步骤4: 选择列 (select, selectExpr)")
print("=" * 80)

# 普通选择
df_combined.select("id", "name").show(3)

# 使用col函数和表达式
df_combined.select(col("id"), col("name"), expr("score * 1.1 as bonus")).show(3)

# 使用selectExpr写SQL表达式
df_combined.selectExpr("id", "name", "score", "CASE WHEN score >= 3000 THEN 'High' ELSE 'Low' END as level").show(3)

# ========== 6. 添加/删除/重命名列 ==========
print("\\n" + "=" * 80)
print("步骤5: 列操作 (withColumn, drop, withColumnRenamed)")
print("=" * 80)

# 添加列：score转换为千分位显示（字符串），增加奖金列
df_with_new = df_combined \\
.withColumn("score_k", col("score") / 1000) \\
.withColumn("bonus", when(col("score") > 3000, col("score") * 0.1).otherwise(col("score") * 0.05)) \\
.withColumn("full_name", concat(col("name"), lit(" (Employee)"))) \\
.drop("score_k") # 演示drop，先加再删

print("添加多列后:")
df_with_new.select("name", "score", "bonus", "full_name").show()

# 重命名列
df_renamed = df_combined.withColumnRenamed("score", "salary")
print("重命名后列名:", df_renamed.columns)

# ========== 7. 过滤数据 ==========
print("\\n" + "=" * 80)
print("步骤6: 过滤 (filter / where)")
print("=" * 80)

# 多种过滤方式
filtered1 = df_combined.filter(col("score") > 2800)
filtered2 = df_combined.where(df_combined.score > 2800)
filtered3 = df_combined.filter("score > 2800")
print("score > 2800 的记录:")
filtered1.show()

# 复合条件：IT部门且score>2500
filtered_complex = df_combined.filter((col("dept") == "IT") & (col("score") > 2500))
print("IT部门且score>2500:")
filtered_complex.show()

# 使用isin
filtered_in = df_combined.filter(col("dept").isin(["IT", "HR"]))
print("部门为IT或HR:")
filtered_in.show(2)

# ========== 8. 去重与采样 ==========
print("\\n" + "=" * 80)
print("步骤7: 去重 (distinct, dropDuplicates) 与采样")
print("=" * 80)

# 创建包含重复数据的DataFrame
dup_data = [(1, "Alice"), (1, "Alice"), (2, "Bob")]
dup_df = spark.createDataFrame(dup_data, ["id", "name"])
print("原始重复数据:")
dup_df.show()
print("distinct去重:")
dup_df.distinct().show()
print("按id去重（保留第一个）:")
dup_df.dropDuplicates(["id"]).show()

# 采样
sample_df = df_combined.sample(fraction=0.5, withReplacement=False, seed=42)
print("50%采样结果（行数可能减少）:")
sample_df.show()

# ========== 9. 排序与限制 ==========
print("\\n" + "=" * 80)
print("步骤8: 排序 (orderBy, sort) 与限制 (limit)")
print("=" * 80)

# 按score降序，然后按name升序
sorted_df = df_combined.orderBy(col("score").desc(), col("name").asc())
print("按score降序排序:")
sorted_df.show()

# 取前3条
limit_df = df_combined.limit(3)
print("前3条:")
limit_df.show()

# ========== 10. 分组聚合 ==========
print("\\n" + "=" * 80)
print("步骤9: 分组聚合 (groupBy, agg)")
print("=" * 80)

# 简单分组计数
dept_count = df_combined.groupBy("dept").count()
print("各部门人数:")
dept_count.show()

# 多种聚合指标
dept_stats = df_combined.groupBy("dept").agg(
avg("score").alias("avg_score"),
spark_sum("score").alias("total_score"),
count("*").alias("cnt")
)
print("各部门统计:")
dept_stats.show()

# 使用pivot（类似Excel透视表）- 需要先准备数据
# 假设有年份部门薪水数据
pivot_data = [(2020, "IT", 1000), (2020, "HR", 800), (2021, "IT", 1200), (2021, "HR", 900)]
pivot_df = spark.createDataFrame(pivot_data, ["year", "dept", "salary"])
pivot_result = pivot_df.groupBy("year").pivot("dept").sum("salary")
print("透视表（每年各部门薪水总和）:")
pivot_result.show()

# ========== 11. 连接操作 (Join) ==========
print("\\n" + "=" * 80)
print("步骤10: 连接DataFrame (join)")
print("=" * 80)

# 创建部门奖金系数表
bonus_df = spark.createDataFrame([("IT", 0.2), ("HR", 0.15), ("Finance", 0.25)], ["dept", "bonus_rate"])
print("部门奖金系数表:")
bonus_df.show()

# 内连接
joined = df_combined.join(bonus_df, on="dept", how="inner")
print("内连接结果:")
joined.select("id", "name", "dept", "score", "bonus_rate").show()

# 左连接
left_joined = df_combined.join(bonus_df, on="dept", how="left")
print("左连接（保留所有员工）:")
left_joined.select("id", "name", "dept", "score", "bonus_rate").show()

# 带条件的连接（非等值连接）
# 例如：分数在某个区间匹配奖金等级（略）

# ========== 12. 处理缺失值 ==========
print("\\n" + "=" * 80)
print("步骤11: 缺失值处理 (dropna, fillna)")
print("=" * 80)

# 创建含有空值的DataFrame
null_data = [(1, "Alice", None), (2, None, 200), (3, "Bob", 300)]
null_df = spark.createDataFrame(null_data, ["id", "name", "score"])
print("原始含空值数据:")
null_df.show()
print("删除任何包含空值的行:")
null_df.dropna().show()
print("删除name列全为空的行（只检查指定列）:")
null_df.dropna(subset=["name"]).show()
print("填充空值:")
null_df.fillna({"name": "Unknown", "score": 0}).show()

# ========== 13. 与RDD互转示例 ==========
print("\\n" + "=" * 80)
print("步骤12: DataFrame <-> RDD 转换")
print("=" * 80)

# DataFrame -> RDD
rdd_converted = df_combined.rdd
print("转换后的RDD分区数:", rdd_converted.getNumPartitions())
print("RDD第一行:", rdd_converted.first())

# RDD -> DataFrame (需指定schema)
new_df = spark.createDataFrame(rdd_converted, schema=df_combined.schema)
print("RDD转回DataFrame后Schema一致:", new_df.schema == df_combined.schema)

# ========== 14. 写入数据到外部存储 ==========
print("\\n" + "=" * 80)
print("步骤13: 写入数据")
print("=" * 80)

output_dir = "/tmp/df_output"
# 写入Parquet（推荐）
df_combined.write.mode("overwrite").parquet(f"{output_dir}/parquet")
print(f"Parquet输出到 {output_dir}/parquet")

# 写入JSON
df_combined.write.mode("overwrite").json(f"{output_dir}/json")
print(f"JSON输出到 {output_dir}/json")

# 写入CSV（需要合并文件，默认会分多部分）
df_combined.coalesce(1).write.mode("overwrite").csv(f"{output_dir}/csv", header=True)
print(f"CSV输出到 {output_dir}/csv")

# 清理临时文件
import shutil
shutil.rmtree(output_dir, ignore_errors=True)

# ========== 15. 性能优化提示 ==========
print("\\n" + "=" * 80)
print("步骤14: 性能优化小贴士")
print("=" * 80)
print("""
– 尽量使用内置函数，避免UDF（Python UDF会破坏Catalyst优化）
– 使用select时只选择需要的列，减少数据传输
– filter尽可能提前，实现谓词下推
– join时广播小表：`df_big.join(broadcast(df_small), "key")`
– 合理设置`spark.sql.shuffle.partitions`（默认200，可根据数据量调整）
– 使用`df.cache()`缓存重复使用的DataFrame
""")

# ========== 16. 清理资源 ==========
spark.stop()
print("\\n✅ DataFrame基础操作演示完成")

八、代码逐行解析

8.1 创建DataFrame部分

• spark.createDataFrame是最核心的方法，第一个参数可以是列表、RDD，第二个参数是schema（可以是列名列表或StructType）。
• 显式定义StructType时，StructField包含名称、类型、是否可为空。可以为复杂类型嵌套。
• 从RDD[Row]创建时，Row对象可以像字典一样访问字段，Spark会自动推断Schema。

8.2 合并DataFrame

• union要求两个DataFrame的Schema完全一致（包括列顺序）。如果不一致，需要用select调整顺序或添加缺失列。
• unionByName可以根据列名合并（Spark 3.x支持），更安全但略慢。

8.3 列操作

• withColumn用于添加或替换列，第一个参数是列名，第二个是列表达式。
• when + otherwise实现条件分支，比在Python中写UDF高效得多。
• drop可以一次删除多列：df.drop("col1", "col2")。

8.4 过滤

• filter和where完全相同，可接受字符串表达式或Column对象。
• 使用&（与）、|（或）、~（非）组合多个条件，注意括号。

8.5 聚合与连接

• groupBy返回GroupedData对象，再调用agg或内置聚合函数。
• pivot是高级功能，用于将行转列，但可能产生大量列，需注意性能。
• join支持inner、left、right、outer等类型，可指定多个连接键：on=["key1", "key2"]。

8.6 缺失值处理

• dropna的thresh参数：至少有多少个非空值才保留行；subset指定检查的列。
• fillna可以传入字典，按列填充不同值。

九、业务场景落地应用

9.1 场景一：数据湖中的批量数据读取

某公司每天将业务数据写入S3的Parquet文件，按日期分区（/data/dt=2024-01-01/）。使用Spark SQL可以高效读取指定日期范围的数据。

df = spark.read.parquet("/data")
df.filter((col("dt") >= "2024-01-01") & (col("dt") <= "2024-01-31")) \\
.groupBy("product_id").agg(sum("revenue")).show()

9.2 场景二：多源数据融合

实时任务需要将Kafka中的用户行为数据（JSON）与MySQL中的用户维度表关联。

# 读取Kafka（结构化流，略），转为DataFrame
# 读取MySQL用户表
users_df = spark.read.format("jdbc").options(
url="jdbc:mysql://db:3306/warehouse",
dbtable="users",
user="spark",
password="xxx"
).load()
# 流式DataFrame与静态DataFrame做lookup join
joined = streaming_df.join(broadcast(users_df), "user_id", "left")

9.3 场景三：数据质量检查

利用dropna、fillna和自定义检查规则，生成数据质量报告。

# 统计每列缺失率
total = df.count()
null_counts = df.select([(count(when(col(c).isNull(), c)) / total).alias(c) for c in df.columns])

9.4 场景四：生成报表并导出

将聚合结果写入Excel（需转Pandas）或数据库。

result_df = df.groupBy("category").sum("sales")
result_df.write.jdbc(url="jdbc:mysql://dw:3306/report", table="daily_sales", mode="overwrite")

十、常见报错排查

10.1 AnalysisException: cannot resolve 'xxx'

原因：引用的列名不存在，或列名大小写写错（Spark默认区分大小写，除非配置spark.sql.caseSensitive=false）。

解决：检查列名，使用df.columns查看。

10.2 TypeError: condition should be string or Column

原因：filter中使用了Python布尔表达式如df.age > 18（df.age是Column对象，但>返回Column，可以这样写，但容易混淆）。实际错误可能是将and、or用于Column对象。

解决：使用&、|运算符，并用括号括住每个条件。

10.3 Py4JError 在collect时

原因：UDF或表达式中有非法操作（如除零），导致Executor端Python进程崩溃。

解决：检查UDF内部异常处理，或用when避免除零等。

10.4 读取CSV时数据全变为null

原因：CSV中某些列的类型无法推断，比如数值列中包含空字符串。或者分隔符错误、转义符问题。

解决：显式指定schema，设置.option("mode", "PERMISSIVE")或.option("nullValue", "")。

10.5 org.apache.spark.SparkException: Job aborted due to stage failure

原因：Shuffle阶段数据倾斜或节点失败。

解决：检查数据是否倾斜，增加分区或使用加盐打散。

十一、本节课知识点总结

DataFrame创建方式一览

常用操作速查

性能最佳实践

十二、课后思考作业

作业一：理论理解题

作业二：代码实践题

作业三：场景应用题

某电商平台每天产生大量订单日志（JSON格式），包含字段：order_id、user_id、amount、status、create_time。现需要每日统计：

• 当天订单总数、总金额、用户数（去重）
• 按小时统计订单量分布
• 筛选出金额大于1000的高价值订单，单独输出

请设计基于DataFrame的方案，写出核心代码，并说明如何优化读取JSON文件的性能。

作业四：拓展研究

提交方式：本次作业要求提交可运行的Python脚本和运行结果截图。鼓励使用真实数据集进行练习。

扩展阅读：

• Spark官方DataFrame API文档：pyspark.sql module
• 《Spark: The Definitive Guide》第8-12章

通过本节课的学习，你已经掌握了DataFrame的创建和基础操作，能够处理大部分日常的数据提取、转换、加载任务。下一节课我们将深入学习PySpark SQL的完整语法，包括过滤、分组、聚合、排序、多表关联等高级查询技巧，让你像操作数据库一样灵活处理分布式数据。我们下节课见！

🔗《20节课 PySpark 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！ 💡 如果本文对您有所启发欢迎： 👍 点赞📌 收藏 📤 分享给更多需要的伙伴。 🗣️ 期待在评论区看到您的想法, 共同进步。 🔔 关注我，持续获取更多干货内容～ 🤗 我们下篇文章见～