企业官网建设流程全解析

简介：面向大数据课程学习者与 Spark 初学者的实验报告文档，围绕 Hadoop 3.1.3 与 Spark 的安装配置、spark-shell 交互使用、本地文件及 HDFS 文件行数统计、Scala 独立应用开发、sbt 打包与 spark-submit 提交运行展开，并包含双文件合并去重、多学科成绩求平均等初级编程实践，以及路径书写错误引发的异常与解决办法。压缩包共 1 个 docx 文件，约 1.9MB，按实验环境、实验内容与完成情况、出现问题、解决方案组织，配有 Windows 与 Ubuntu Kylin 环境说明、操作截图、关键代码和运行结果，实验记录较完整，从环境搭建到应用提交形成闭环，便于读者对照复现实验流程、核对输出结果、理解 Spark 基本数据操作与独立应用部署。已有 8348 人学习下载，适合课程实验参考、期末复盘和入门阶段查漏补缺，也可作为实验报告撰写范例与流程梳理参考。

1. 一份几万行日志引发的疑问：Spark 初级编程实践到底在练什么

某高校的数据处理课把实验七定成「Spark 初级编程实践」，题目本身很朴素：给一份几万行的访问日志，统计词频、按字段分组求均值、取出访问量最高的前十条。很多人第一反应是写 Python 循环加字典，本地跑几十秒也能出结果，于是觉得没必要上 Spark。真正卡人的在后面——数据涨到几百万行，或者要求同一份代码不做改动就能扩到多台机器上跑，纯 Python 的写法就得推倒重来。Spark 初级编程实践要建立的能力，是把「循环处理每一条数据」换成「描述对一整个数据集做什么变换和聚合」，用 RDD 和 DataFrame 两套接口把统计题写出来，并且在单机上跑通、看清每一步的执行过程。适合刚接触分布式计算、手上只有一台笔记本、还得交出可复现结果的人。

2. 先把能跑的最小环境搭起来：local 模式、Java 依赖与第一个 SparkSession

环境是最容易劝退人的一关。很多人第一次装完 PySpark，敲下pyspark看到满屏 WARN 就慌了，其实大部分警告不影响计算。这一章要达成的是：一条命令能启动，一段代码能跑出正确数字，知道每个参数改了会带来什么变化。

2.1 课程实验为什么优先用 local[*] 而不是先搭集群

实验阶段的头号误区是「不上集群就不算分布式」。事实上 Spark 的编程模型和执行语义在单机上完全一致，local[*]会在同一个 JVM 里起多个线程模拟并行，你的map、reduceByKey、repartition该触发 shuffle 照样触发，explain里该出现Exchange照样出现。反倒是硬搭 Standalone 集群，多出来的是一堆和实验目标无关的麻烦：主节点和从节点要能互相解析主机名，worker 要注册到 master，端口要放通，Driver 提交上去以后本地文件路径读不到。

我一般给学员的建议是：实验一到七全部在 local 模式下做完，把master理解成一个可替换的参数就够了。真正需要集群的场景是数据量超单机内存或者要验证资源调度，那已经不是「初级编程实践」的范围。

local[4]比local[*]更适合写进实验报告，因为换台机器跑出来分区数不变，截图和结论能对得上。local[*]在你的电脑上并行度可能是 8，在机房机器上可能是 16，getNumPartitions打出来不一样，容易被判成「结果不可复现」。

2.2 装 PySpark 之前，先过 Java 这一关

PySpark 只是 Spark 的 Python 外壳，真正的计算引擎跑在 JVM 上，所以机器上必须有可用的 Java，并且JAVA_HOME要指向 JDK 的根目录。这一步翻车的人最多，因为报错信息看起来和 Python 毫无关系。

# 1) 确认 Java 存在，且输出里带 version 字样 java -version # 2) 建一个独立虚拟环境，避免和系统里已有的 pyspark 打架 python -m venv spark-lab source spark-lab/bin/activate # Windows 用 spark-lab\Scripts\activate pip install pyspark # 3) 确认命令入口和版本，能打印出版本号就算装好了 pyspark --version

Java 已装但JAVA_HOME没配时，可以这样临时补上。Linux 和 macOS 下：

# readlink -f 拿到 java 可执行文件的真实路径，再往上退两级就是 JDK 根目录 export JAVA_HOME=$(dirname $(dirname $(readlink -f $(which java)))) export PATH=$JAVA_HOME/bin:$PATH

Windows 下的细节不一样：JAVA_HOME要设成 JDK 安装目录本身，比如C:\Program Files\Java\jdk-17，不要把\bin带进去。设错成...\bin是典型错误，症状是启动时报Error: Could not find or load main class，看着像 classpath 问题，实际是路径多了一层。

提示：把这几个export写进 shell 配置文件，或者在项目里用一个env.sh统一 source，比每次开终端手敲靠谱得多。

参数层面只有两个值得记：pip install pyspark装的是完整包，自带spark-submit、pyspark命令行入口和示例 jar，几百 MB 属于正常；虚拟环境激活后再装，pyspark命令才会落在当前环境里，否则你pip install到一半切了终端，用的时候又会找不到命令。

2.3 第一个 SparkSession：用三行代码确认环境真的通了

环境装好不等于能跑。最快的确诊方式是跑一个不读文件、不依赖外部资源的最小脚本，把「JVM 能起来」「Python worker 能通信」「行动算子能返回结果」三件事一次性验证掉。

from pyspark.sql import SparkSession # master=local[*] 用满本机核心；appName 只影响日志和 UI 里的标识 # shuffle.partitions 默认 200，单机实验里没必要开这么多，降到 8 减少小文件 spark = (SparkSession.builder .master("local[*]") .appName("lab7-smoke-test") .config("spark.ui.port", "4040") .config("spark.sql.shuffle.partitions", "8") .getOrCreate()) sc = spark.sparkContext sc.setLogLevel("WARN") # 默认 INFO 太吵，实验阶段调成 WARN rdd = sc.parallelize(range(1, 101), 4) print("分区数:", rdd.getNumPartitions()) print("求和:", rdd.reduce(lambda a, b: a + b)) print("前 5 个:", rdd.take(5)) spark.stop()

逻辑说明：SparkSession.builder是唯一的入口，在 PySpark 里它同时代理了老的SparkContext和SQLContext，所以只需要建一次。getOrCreate()而不是builder.getOrCreate之外的新建，是因为同一个 JVM 里重复创建会抛异常，用getOrCreate可以复用。parallelize(数据, 4)的第二个参数是分区数，reduce和take都是行动算子，它们返回的是 Python 原生对象，能直接print。

参数说明：spark.ui.port指定 Web UI 端口，被占用时 Spark 会自动往后顺延，日志里会打实际端口；spark.sql.shuffle.partitions只对 DataFrame/SQL 的 shuffle 生效，RDD 的reduceByKey走的是另一个参数spark.default.parallelism，这一点很多人混。看到WARN NativeCodeLoader: Unable to load native-hadoop library这类警告不用管，它表示没有本地压缩库，功能不受影响。

3. RDD 的三件事：怎么创建、变换为什么不执行、聚合该选哪个算子

RDD 是 Spark 最底层的抽象，理解它之后再看 DataFrame 就是「带 schema 的 RDD 加一个优化器」。这一章把创建、转换、行动三类操作拆开，重点讲清楚延迟计算带来的误解，以及词频统计里reduceByKey和groupByKey的真实差别。

3.1 parallelize 与 textFile：两种来源的分区数由谁决定

分区数直接决定并行度，也决定后面 shuffle 的开销，初级实验里最值得先观察的就是它。

from pyspark.sql import SparkSession spark = SparkSession.builder.master("local[4]").appName("lab7-rdd").getOrCreate() sc = spark.sparkContext sc.setLogLevel("WARN") # 来源一：本地集合，分区数由第二个参数直接指定，不给就用默认并行度 rdd1 = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8], 2) print("parallelize 分区数:", rdd1.getNumPartitions()) # 2 # 来源二：文本文件，按行读，minPartitions 只是建议下限 rdd2 = sc.textFile("data/access.log", minPartitions=2) print("textFile 分区数:", rdd2.getNumPartitions()) print("默认并行度:", sc.defaultParallelism) # local[4] 下是 4 # 查看每个分区里到底装了什么，用来验证分区不是均分的 print(rdd1.glom().collect())

逻辑说明：parallelize把 driver 端的一个 Python 列表切成 N 份发到各分区，数据量小时分区之间可能很不均匀，glom().collect()能把它打印出来。textFile走的是 Hadoop 的输入格式，分区数由文件的块数量和minPartitions共同决定——文件只有一个、又比一个块小的时候，即使你写了minPartitions=8，实际分区数也可能是 1。

参数说明：sc.defaultParallelism在local[N]下等于 N，在local[*]下等于本机核心数，它是reduceByKey、join这类算子的默认分区依据。textFile的第二个位置参数就是minPartitions，写 2 表示「至少切两块」，但如果文件本身只占一个块，Spark 不会为了凑数去硬切，这是新手最容易误判的一点。

3.2 转换与行动：为什么 map 写完什么都没发生

Spark 的转换算子全是延迟的，它们只往血缘图里追加一个节点，不提交任务。真正触发计算的是行动算子。

lines = sc.textFile("data/access.log") # 下面这一串全是转换算子，写完这一行程序不会有任何输出 parsed = (lines.filter(lambda l: l.strip() != "") .map(lambda l: l.split("\t")) .filter(lambda f: len(f) >= 4)) # 直接打印 RDD 对象，只会看到类似 PythonRDD[12] at RDD at PythonRDD.scala 的字符串 print(parsed) # 这三个是行动算子，才会把任务提交到执行器 print("总行数:", parsed.count()) print("前 3 条:", parsed.take(3)) print("抽样 2 条:", parsed.takeSample(False, 2, seed=42))

逻辑说明：print(parsed)打出的是 RDD 的元信息，不是数据。第一次跑实验的人常把这一行当成验证，以为代码没生效，其实是自己没触发计算。count()会把每个分区的计数汇总回 driver；take(3)只扫描需要的分区，命中 3 条就停，比collect()安全得多；takeSample带seed时可以复现，写实验报告时很有用。

参数说明：take(n)的 n 要小，它会把结果拉回 driver 内存；collect()在数据量大时会直接把 driver 内存占满导致java.lang.OutOfMemoryError，本地跑几百万行日志时尤其要注意。filter里判断len(f) >= 4是防止后面按列取值时IndexError，脏数据在日志文件里是常态，不是例外。

3.3 词频统计的两种写法：reduceByKey 与 groupByKey 差在哪

同一道题两种写法都能出结果，但执行过程完全不同，这是初级实验里最该讲清楚的一处。

words = sc.textFile("data/words.txt").flatMap(lambda line: line.split()) # 写法 A：reduceByKey，map 端先做局部合并，再跨分区合并 counts_a = words.map(lambda w: (w, 1)).reduceByKey(lambda a, b: a + b) # 写法 B：groupByKey，把同一个 key 的所有 value 全拉到同一分区再求和 counts_b = words.map(lambda w: (w, 1)).groupByKey().mapValues(sum) # 取词频 Top10；takeOrdered 在 driver 端只保留 n 条，比 sortBy 全排序省内存 top10 = counts_a.takeOrdered(10, key=lambda kv: -kv[1]) for w, c in top10: print(w, c) # 两种写法的物理计划对比，重点看有没有 Exchange counts_a.explain()

逻辑说明：flatMap把一行拆成多个词，返回的是「扁平」的元素序列而不是列表的列表，所以后面可以直接map成(词, 1)。写法 A 的reduceByKey会在 map 端先做一次局部聚合，网络上传的是「局部结果」；写法 B 的groupByKey不做局部聚合，同一个 key 的所有1都要经过网络，数据倾斜时某个分区会撑爆。

参数说明：reduceByKey(func, numPartitions=None)的第二个参数不给就用defaultParallelism，显式写 4 或 8 能让结果分区数稳定。takeOrdered(n, key=...)的 key 用-kv[1]表示按次数降序；等价的sortBy(lambda kv: kv[1], ascending=False).take(10)会对全量数据排序，数据量大时慢得多。

4. 把统计题改写成 DataFrame：schema、临时视图与结果落盘

DataFrame 让 Spark 能做列裁剪和谓词下推，同样的统计题写起来更短、跑起来往往更快。初级实验里通常要求「用 SQL 完成一次分组统计并输出结果文件」，这一章按这条线走一遍。

4.1 RDD 转 DataFrame：为什么建议显式声明 schema

用inferSchema=True读 CSV 很方便，但它要多扫一遍数据，而且把整数推断成字符串、把空值推成null的情况很常见。实验数据字段固定，直接写StructType更稳。

from pyspark.sql import SparkSession from pyspark.sql.types import (StructType, StructField, StringType, IntegerType) spark = SparkSession.builder.master("local[4]").appName("lab7-sql").getOrCreate() spark.sparkContext.setLogLevel("WARN") # 字段名、类型、是否允许为空，三者都要显式给出 schema = StructType([ StructField("user_id", StringType(), False), StructField("page", StringType(), False), StructField("stay_sec", IntegerType(), True), ]) rows = [("u001", "home", 12), ("u001", "detail", 35), ("u002", "home", 8), ("u002", "detail", 21), ("u003", "home", 5)] df = spark.createDataFrame(rows, schema=schema) df.printSchema() df.show(5, truncate=False)

逻辑说明：createDataFrame(数据, schema)的第二个参数可以是StructType，也可以是 DDL 字符串如"user_id STRING, page STRING, stay_sec INT"，后者写起来短，但字段一多容易漏逗号。printSchema()打印树形结构，是确认类型有没有推错的最快方式；show(n, truncate=False)不截断长字符串，检查内容时比默认的show()好用。

参数说明：StructField的第三个参数nullable设成False后，如果数据里真有None会直接报错，这在清洗阶段反而是好事——能早点发现脏数据。从已有 RDD 转过来时用spark.createDataFrame(rdd, schema)，要求 RDD 的每个元素是 tuple 或 Row，长度和字段数必须对齐，否则报Length of object does not match。

4.2 注册临时视图，用 SQL 做分组与 TopN

临时视图只存在于当前 SparkSession 的生命周期里，进程一停就没了，正好符合实验场景。

df.createOrReplaceTempView("pv_log") result = spark.sql(""" SELECT user_id, COUNT(*) AS pv, ROUND(AVG(stay_sec), 2) AS avg_stay FROM pv_log WHERE stay_sec IS NOT NULL GROUP BY user_id ORDER BY pv DESC, avg_stay DESC """) result.show(10, truncate=False)

逻辑说明：createOrReplaceTempView注册的是会话级视图，重复执行不会报「已存在」。WHERE要在GROUP BY之前过滤掉空值，否则AVG会自动忽略null但COUNT(*)不会，两个指标口径就不一致了——这是统计题里最常见的口径错误。ORDER BY里第二个排序键做兜底，保证 pv 相同时输出顺序稳定，截图不会每次都变。

参数说明：ROUND(x, 2)保留两位小数，纯展示用，不影响后续计算。真的要在代码里继续算，就别ROUND，保留原始精度到最后一步再格式化。spark.sql.shuffle.partitions在这里起作用：GROUP BY会触发一次 shuffle，默认 200 个分区在单机上会生成一堆小任务，启动开销比计算还大，前面把它设成 8 就是这个原因。

4.3 结果落盘：coalesce(1)、覆盖模式与 part 文件

实验报告一般要求提交结果文件，直接write会遇到两个坑：目录已存在报错，以及输出成一堆part-00000小文件。

# 输出目录必须不存在，否则报 AnalysisException: path already exists # mode("overwrite") 省掉手动删目录这一步；coalesce(1) 把结果合并到一个分区 (result.coalesce(1) .write .mode("overwrite") .option("header", "true") .csv("out/user_pv"))

逻辑说明：coalesce(1)是窄依赖，只做分区合并、不触发 shuffle，所以比repartition(1)便宜；代价是如果上游数据很大，单个分区会撑不住，实验结果通常只有几十行，不存在这个问题。写完之后目录里会出现part-00000-xxxx.csv和_SUCCESS，前者是数据，后者是个空标记文件，用来表示这次写入完整成功。

参数说明：mode有overwrite、append、ignore、error四种，实验里用overwrite最省事；option("header","true")只对 csv 生效，写 parquet 时不需要；要指定分隔符再加.option("sep", "\t")。_SUCCESS文件不要删，很多下游流程靠它判断数据是否就绪。

5. 五个反复出现的翻车现场：从报错信息反推原因

这一章按「现象 → 原因 → 解决」记录五个在实验里高频出现的故障，都是我从别人的报错截图里反复见到的同一批问题。

5.1 启动就报 Java gateway process exited

现象：python test.py一运行就抛Exception: Java gateway process exited before sending its port number，Python 侧连一行自己的日志都没打出来。

原因：JAVA_HOME没设、设错，或者指向了一个不完整的 JRE。PySpark 启动时要拉起一个 Java 网关进程，靠JAVA_HOME定位java可执行文件，找不到就直接退出，而 Python 侧只能看到一个笼统的异常。

解决：先在同一个终端里执行echo $JAVA_HOME和java -version，两条都能正常输出再跑脚本。Windows 上确认路径没有多余的\bin，路径里有空格时不要自己加引号，交给系统处理。如果装了多个 JDK，用which java确认当前生效的那个和JAVA_HOME指向同一个。

5.2 Python worker crashed，或者执行器里的 Python 不是同一个

现象：小数据量跑得好好的，数据一多就报Python worker exited unexpectedly (crashed)，日志里还带着一串看不懂的 native 栈。

原因：driver 用的 Python 和 executor 拉起 worker 用的 Python 不是同一个解释器，尤其常见于虚拟环境里跑脚本、但环境变量没传下去的情况。两边的 Python 版本或第三方库版本不一致时，序列化过去的闭包在 worker 端反序列化就会崩。

解决：显式指定解释器路径，让 driver 和 worker 用同一个。

# 两个变量都指向当前虚拟环境的 python，避免用系统自带的那个 export PYSPARK_PYTHON=$(which python) export PYSPARK_DRIVER_PYTHON=$(which python)

写完再跑一次python -c "import sys; print(sys.executable)"核对路径。另外，闭包里不要引用不可序列化的对象，比如打开的文件句柄、数据库连接、SparkSession 本身，这些都会在 worker 端引发同类崩溃。

5.3 写结果时报 path already exists

现象：第一次write.csv("out/result")成功，改完代码再跑就抛AnalysisException: path file:.../out/result already exists。

原因：Spark 的写入语义是「目录不存在才写」，它不会默默覆盖已有目录，这是为了防止误删数据。这和很多人习惯的open(..., "w")覆盖写完全不同。

解决：写的时候加.mode("overwrite")，或者在脚本里先用shutil.rmtree("out/result", ignore_errors=True)清掉。调试阶段我倾向用后者，因为删的动作显式可见；正式提交的代码用overwrite，少一次文件系统操作。

5.4 groupByKey 跑着跑着 OOM

现象：词频统计用groupByKey写法，数据量一大就报java.lang.OutOfMemoryError: Java heap space，或者某个 task 卡在 99% 不动。

原因：groupByKey不做事先合并，同一个 key 的所有 value 会被完整地拉到同一个分区，遇到热点 key（比如日志里反复出现的空字符串或某个固定 URL）时，单个分区的数据量能压垮堆内存。

解决：换成reduceByKey或aggregateByKey，让 map 端先局部聚合。如果业务上确实需要拿到每个 key 的全部明细，用combineByKey自定义聚合逻辑，或者先给 key 加随机前缀打散、聚合两次，再合并结果。判断依据很简单：explain()里看到Exchange hashpartitioning后面跟着的数据量远大于输入，就是倾斜。

5.5 Windows 下写文件报 NullPointerException

现象：Linux 上跑得好好的代码，在 Windows 上做write时抛java.lang.NullPointerException，栈里出现NativeIO$Windows字样。

原因：Hadoop 在 Windows 上需要额外的本地库才能做文件系统操作，缺少时会在这类调用上返回空指针。

解决：设好HADOOP_HOME指向一个包含bin目录的路径，把对应版本的winutils.exe和hadoop.dll放进去，再把%HADOOP_HOME%\bin加进PATH，重启终端。更省事的做法是实验阶段直接输出到本地路径并改用mode("overwrite")，或者干脆在 Linux 环境/容器里做——为了一个输出动作去折腾本地库，投入产出比不高。

6. 让结果从「能跑」变成「看得出门道」：cache 与执行计划

实验报告里最容易失分的一句话是「程序运行成功」。评委想看的是你知不知道自己省了多少计算。同一份 RDD 被多个行动算子复用而不加缓存，每次都要从头重算整条血缘，这是最隐蔽的浪费。我一般的习惯是：只要一个中间结果会被用两次以上，就先cache()，第一次行动算子触发物化，后续直接命中内存。explain(True)则用来确认GROUP BY那一步到底有没有产生Exchange——没有Exchange说明走的是窄依赖，有的话就该想想分区数设成多少合适。

counts = words.map(lambda w: (w, 1)).reduceByKey(lambda a, b: a + b) counts.cache() # 只标记，不计算 counts.count() # 第一个行动算子，此时才真正物化到内存 counts.take(10) # 第二次复用，不再重算整条血缘 counts.explain(True) # 看物理计划里的 Exchange 和 Scan

有个细节要留意：cache()之后如果马上unpersist()，缓存等于白做；数据量超过可用内存时，Spark 会把部分分区溢出到磁盘，性能反而可能不如不缓存，这时候用persist(StorageLevel.MEMORY_AND_DISK)更稳。我在第一次带人做这个实验时，就是因为没看explain，把 200 个 shuffle 分区当成默认最优，结果单机跑了七分钟才出结果，降到 8 之后二十秒就完了——参数不是越默认越安全，得看你跑在什么规模的机器上。希望帮到你。

本文还有配套的精品资源，点击获取