LabEx
免费加入

理解 Spark 执行与常见错误

大数据Beginner
立即练习

介绍

你已经编写 Spark 转换、触发行动操作并保存数据。现在将把这些熟悉操作与执行证据联系起来：描述工作的计划，以及记录作业、阶段和任务的真实事件日志。目标是解释行为、读懂有用错误，不是记忆内部任务数。

driver（驱动程序）运行应用并协调工作。在集群中，执行器在分配的资源上运行任务；local[2] 则在当前机器上用两个执行线程运行 Spark 工作。作业可以划分为多个阶段，每个阶段针对计算分区执行任务。一次行动操作可能触发多个作业，尤其涉及查询规划和自适应行为时；具体数量可能变化。

本独立实验提供干净订单、客户、两个故意写错的表达式，以及只读事件汇总工具。你应该了解筛选、分组、关联和 Parquet 输出。本实验不调优资源、不将一般数据集收集到 Python 循环，也不部署集群。请使用终端。

将熟悉查询与执行计划联系起来

在这一步中，你将在请求结果前查看每日地区已付款销售查询。读取器、筛选、金额、客户关联和聚合均重复之前技能。

这里的 groupBy("order_date", "region") 同时使用两个键：每个不同的日期与地区组合形成一组。10 月 14 日的 East 与 10 月 15 日的 East、10 月 14 日的 West 都是独立组。计数和求和在每个组合内进行。这将之前的每日与地区问题合并为一份报表。explain(mode="formatted") 打印计划算子，不返回报表行。

SparkSession 还会启用 output/spark-events 中的真实事件日志。关闭压缩是为了应用停止后能够阅读教学日志。这些设置记录执行，不用伪造作业记录代替。目标目录已为你准备好。

开始并运行完整检查脚本：

cd /home/labex/project

cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = (
    SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
    .config("spark.eventLog.enabled", "true")
    .config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
    .config("spark.eventLog.compress", "false")
    .config("spark.eventLog.rolling.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
    col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
    count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt

cat /home/labex/project/output/execution-plan-and-result.txt

在 Report execution plan 中，查找筛选、投影、关联、聚合的作用，以及在计算分区之间移动记录的 Exchange。名称和计划结构可能随优化而变化。Exchange 是执行边界，不是新输出目录。

Report action result 下预期为 10 月 14 日 / East / 1 / 20.00，10 月 14 日 / West / 1 / 25.00，10 月 15 日 / East / 1 / 50.00，以及 10 月 15 日 / West / 1 / 30.00。总计仍为 125.00。show 和写入器请求实际执行所描述工作；规划和输入元数据处理也可能进行准备工作。

最后的显示使用 spark_partition_id，列出这些源记录所处的计算分区 ID。dropDuplicates 让每个观察到的 ID 只列一次。ID 和观察到的数量可能变化，空分区不一定出现。这些运行时分组不同于前一个实验中 partitionBy 写出的日期目录。一个组织工作，另一个布局存储数据。

阅读真实作业、阶段和任务事件

在这一步中，你将查看刚运行应用产生的事件日志。提供的工具从目录读取 JSON 事件记录，并汇总实际事件。无需编写日志解析器，也无需选择版本特定文件名。

运行汇总工具，将事件目录作为输入参数。Shell 的 > 保留输出供审核：

bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt

cat /home/labex/project/output/event-summary.txt

预期看到实际应用身份、正数的 Jobs started、Stages submitted 和 Tasks ended 计数，以及作业到阶段的引用。数字取决于本次运行，可能随优化、文件切分和 Spark 版本改变。它们是观察证据，不是需要精确复现的目标。

行动操作请求答案；Spark 可能把所需工作组织为多个作业。作业引用阶段，阶段包含针对工作分区的任务。shuffle 边界可以分隔阶段，多个转换则可在同一阶段内合并。local[2] 限制同时执行的线程，不代表每个应用恰好有两个作业、阶段或任务。

对于微小数据，JVM 启动、调度和文件处理相对于计算本身可能很明显。小幅耗时差异不能形成可靠调优结论。应先通过计划和事件证据理解工作。

诊断并修复未知列

在这一步中，你将先阅读故意触发的名称解析错误，再修正原因。查看预置错误示例：

cat /home/labex/project/tools/broken-column.py

模式声明的是 quantity，最后的 select 却请求 quanity。保存实际失败输出。> 将普通输出重定向到日志，2>&1 将标准错误发到相同目标，因此调用栈信息也会保存。下一条命令预期失败。

bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1

tail -n 20 /home/labex/project/output/column-error.log

找到无法解析的列名，以及可能建议的有效名称。很长的调用栈包含实现细节，但列引用指出了修正方向。不需要重装输入相关工具，也不需要转成 Python 循环。

使用声明的列名创建修正程序，保留错误教学样例供比较：

cat > /home/labex/project/fixed_column.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON

bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt

cat /home/labex/project/output/fixed-column-result.txt

预期有全部六个 ID，数量依次为 2、1、1、2、1、2。投影现在能按真实模式解析，并保存实际结果。列名及其拼写是数据约定的一部分。

修复不是布尔条件的筛选

在这一步中，你将区分有效整数列与有效筛选条件。查看第二个预置示例：

cat /home/labex/project/tools/broken-filter.py

filter 要求每行对应一个布尔表达式。单独的 col("quantity") 是整数，不是真/假判断。保存故意触发的类型不匹配。下一条命令预期失败。

bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1

tail -n 20 /home/labex/project/output/filter-error.log

找到筛选条件及其非布尔整数类型。修复方法是使用已学列运算符表达预期比较 quantity > 0。运行修正后的完整程序：

cat > /home/labex/project/fixed_filter.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON

bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt

cat /home/labex/project/output/fixed-filter-result.txt

六条干净样本行数量均为正，因此全部保留。负数或零数量不满足条件；这项特定修复不应用独立的已付款销售规则。

现在你拥有三个层面的证据：预期表达式和模式、实际计划与执行工作，以及局部修复后的正确持久结果。解释执行时，请保持存储日期目录与计算分区的区别。接下来，你将使用明确输入和输出参数提交独立批处理脚本。

总结

你将熟悉转换与真实计划、事件日志联系起来，观察应用、作业、阶段和任务证据，不固定其数量，并区分计算分区与日期目录。你利用模式和错误证据修复了无法解析的列及非布尔筛选。接下来，明确提交参数将让批处理应用独立于之前进程。

相关 Big Data 课程
Hadoop 入门

Hadoop 入门

bigdata
Hive 入门

Hive 入门

bigdata
Apache Kafka 入门

Apache Kafka 入门

pythonbigdata