简介本资源为大数据课程实验七「Spark初级编程实践」的完整实验报告文档面向正在学习Spark与Hadoop的高校学生及大数据入门开发者帮助解决环境搭建、数据读取与独立应用开发中的实操难题。压缩包内仅含1个docx文件约1.9MB以图文并茂的实验报告形式呈现涵盖实验环境配置、操作步骤截图与代码说明。内容围绕Spark Shell启动、本地与HDFS文件行数统计、Scala独立应用打包提交、数据去重及求平均值等典型任务展开并记录了IllegalArgumentException、InvalidInputException、URISyntaxException等常见报错的成因与修正方法如file:///路径少写斜杠、HDFS根目录应为/user/hadoop、URL前多余空格等排错思路。目前已有8339人学习下载适合需要对照实验流程、复盘代码细节与查漏补缺的初学者参考。1. 实验七Spark初级编程实践到底在练什么很多人第一次看到“实验七Spark初级编程实践”这个标题会下意识觉得它只是一次普通的课程作业——装个环境、跑个 WordCount、截图交差。但真正做过的人知道这个实验的核心不是“跑通”而是让你第一次完整地经历数据从内存到分布式算子的流转过程RDD 怎么切分、惰性求值什么时候触发、shuffle 为什么会让任务突然变慢。这些概念在文档里读十遍不如亲手在本地模式跑一遍再改成集群模式来得深刻。这篇笔记面向两类人一类是正在做 Spark 入门实验、卡在环境或算子语义上的同学另一类是想从零搭一套可复现的 Spark 练手环境、但不想被各种版本冲突折腾的开发者。我会按“先跑通最小闭环再拆算子语义最后处理踩坑”的顺序展开所有代码都可以直接复制到本地或实验集群执行。需要说明的是不同实验环境给的集群规模、Spark 版本可能不同文中参数以本地模式和常见伪分布式为准你按自己环境微调即可。2. 把 Spark 本地环境跑起来从零到第一个 RDD2.1 选本地模式还是集群模式先别急着上集群做 Spark 初级实验最容易翻车的地方是一上来就去连实验室的集群。集群往往有排队、权限、资源池限制一个环境问题能卡掉两小时而你真正要练的算子语义还没碰。我的建议是所有初级实验先在本地模式local[*]跑通再迁移到集群。本地模式用你机器的多核模拟并行RDD、DataFrame、shuffle 行为跟集群基本一致唯一区别是数据量和容错路径。选型上如果你只是做实验七这类初级编程实践本地模式足够覆盖 90% 的算子练习。需要验证分布式特性时再用伪分布式Standalone 或 local-cluster补。下面给一个不依赖任何平台安装包的最小环境方案用 Python 的 venv 隔离避免污染系统环境。# 创建独立环境避免和系统里的 Python 包冲突 python3 -m venv spark-lab source spark-lab/bin/activate # 安装 Spark 运行所需的 Python 侧依赖 pip install pyspark3.5.0 # 验证安装能打印版本即成功 python -c import pyspark; print(pyspark.__version__)这里固定pyspark3.5.0是为了让后续代码的 API 行为可预期。PySpark 的版本和 Scala 侧 Spark 是绑定的如果你机器上已经有 Spark 安装包注意SPARK_HOME指向的版本要和 pip 装的一致否则会出现“Python 端能 import、提交任务却找不到类”的玄学问题。参数上local[*]里的*表示用满所有可用核实验阶段建议显式写成local[2]或local[4]方便观察并行度对任务的影响。2.2 第一个 RDD用 parallelize 和 textFile 建立直觉环境好了之后第一个要建立的直觉是RDD 是“描述”而不是“数据”。下面这段代码同时演示两种创建方式并打印分区数让你看到惰性求值的边界。from pyspark import SparkContext, SparkConf # 显式指定 local[2]方便观察 2 个分区的行为 conf SparkConf().setAppName(lab7-basic).setMaster(local[2]) sc SparkContext(confconf) # 方式一从内存集合创建第二个参数指定分区数 rdd_from_list sc.parallelize([1, 2, 3, 4, 5, 6], 2) print(parallelize 分区数:, rdd_from_list.getNumPartitions()) # 方式二从本地文件创建分区数默认等于文件分块数 rdd_from_file sc.textFile(data/words.txt, minPartitions2) print(textFile 分区数:, rdd_from_file.getNumPartitions()) # 注意到这里为止文件其实还没被真正读取 # 只有遇到 action 算子才会触发计算 print(前 3 个元素:, rdd_from_list.take(3)) sc.stop()逻辑说明parallelize把本地集合切成指定数量的分区适合做小规模算子验证textFile按行读取minPartitions控制最小分区数实际分区可能更多。关键点是take(3)之前的所有print都不会触发文件读取这就是惰性求值——transform 算子只记录血缘action 才真正执行。参数上minPartitions设太小会导致单分区数据倾斜设太大则任务调度开销上升实验阶段 2 到 4 比较合适。2.3 用 WordCount 串起 map、flatMap、reduceByKeyWordCount 是 Spark 的“Hello World”但很多人只是抄了一遍没搞清flatMap和map的区别。下面这版加了中间结果打印帮你看到每一步的数据形态。from pyspark import SparkContext, SparkConf conf SparkConf().setAppName(lab7-wordcount).setMaster(local[2]) sc SparkContext(confconf) lines sc.textFile(data/words.txt) # flatMap一行变多行拆词map一行变一行 words lines.flatMap(lambda line: line.split( )) print(拆词后前 5 个:, words.take(5)) # map 成 (word, 1) 的键值对 pairs words.map(lambda w: (w, 1)) # reduceByKey 会触发 shuffle按 key 聚合 counts pairs.reduceByKey(lambda a, b: a b) # 按词频降序take 是 action result counts.sortBy(lambda x: x[1], ascendingFalse).take(10) for word, cnt in result: print(word, cnt) sc.stop()逻辑说明flatMap的返回是“可迭代的多个元素”所以一行文本能拆成多个词map是一进一出。reduceByKey是宽依赖算子会在 map 端先做本地聚合combiner再 shuffle这比groupByKey高效得多——初级实验里最常见的性能对比就是这两个。参数上sortBy的第二个参数控制升降序take(10)只拉回 10 条到 driver避免把全量结果拉到本地导致内存溢出。3. 算子语义拆解transform 和 action 的边界在哪3.1 为什么你的 RDD 打印出来是空的初学者最常遇到的“翻车”场景写了一段 transform 代码直接print(rdd)结果输出一行PythonRDD[...]或者干脆什么都没有。原因就是 transform 算子不触发计算。下面用对比代码说明。from pyspark import SparkContext, SparkConf conf SparkConf().setAppName(lab7-lazy).setMaster(local[2]) sc SparkContext(confconf) rdd sc.parallelize([1, 2, 3, 4, 5]) # 错误示范只做 transform不触发计算 mapped rdd.map(lambda x: x * 2) print(直接打印 mapped:, mapped) # 输出的是 RDD 对象描述不是数据 # 正确做法用 action 算子触发 print(collect 结果:, mapped.collect()) # 全量拉回 print(count 结果:, mapped.count()) # 只返回计数 print(reduce 结果:, mapped.reduce(lambda a, b: a b)) sc.stop()逻辑说明map只是记录“要对每个元素做乘 2”这个血缘关系collect、count、reduce才是 action。collect会把所有分区数据拉到 driver数据量大时慎用count只返回一个数字适合做验证reduce做聚合。参数上没有额外参数但要注意collect在数据超过 driver 内存时会 OOM实验阶段数据小无所谓真实场景要用take或写回存储。3.2 groupByKey 和 reduceByKey 的性能差异怎么验证这两个算子都能做分组聚合但 shuffle 行为完全不同。下面用同一份数据分别跑观察执行时间差异。import time from pyspark import SparkContext, SparkConf conf SparkConf().setAppName(lab7-shuffle).setMaster(local[2]) sc SparkContext(confconf) # 构造 10 万条键值对key 只有 10 种制造聚合压力 data [(i % 10, i) for i in range(100000)] rdd sc.parallelize(data, 4) # 方式一groupByKey所有 value 先 shuffle 再聚合 t1 time.time() gbk rdd.groupByKey().mapValues(lambda vs: sum(vs)).collect() t2 time.time() print(groupByKey 耗时: %.3f 秒 % (t2 - t1)) # 方式二reduceByKeymap 端先做本地聚合再 shuffle t3 time.time() rbk rdd.reduceByKey(lambda a, b: a b).collect() t4 time.time() print(reduceByKey 耗时: %.3f 秒 % (t4 - t3)) sc.stop()逻辑说明groupByKey会把同一个 key 的所有 value 原样 shuffle 到同一个分区网络和内存开销大reduceByKey在 map 端先用 combiner 做本地聚合shuffle 的数据量大幅减少。参数上parallelize的第二个参数 4 表示 4 个分区分区数越多并行度越高但 shuffle 的文件数也越多。实验里如果数据量太小两者差异不明显把数据量加到百万级就能看到差距。3.3 持久化什么时候该 cache什么时候不该迭代计算场景比如机器学习里的反复扫描如果不做持久化每次 action 都会从头重算整条血缘链。下面演示 cache 的效果。from pyspark import SparkContext, SparkConf, StorageLevel conf SparkConf().setAppName(lab7-cache).setMaster(local[2]) sc SparkContext(confconf) rdd sc.parallelize(range(1000000), 4).map(lambda x: x * x) # 第一次 action触发计算 print(第一次 count:, rdd.count()) # 不做持久化第二次 action 会重新计算整条链 print(第二次 count:, rdd.count()) # 加上 cache第二次直接读缓存 rdd.cache() print(cache 后第一次 count:, rdd.count()) print(cache 后第二次 count:, rdd.count()) # 查看是否真的缓存了 print(是否已缓存:, rdd.is_cached) sc.stop()逻辑说明cache()是persist(StorageLevel.MEMORY_ONLY)的简写把 RDD 数据留在内存。第一次 action 后缓存才生效第二次 action 直接命中。参数上如果内存不够可以用persist(StorageLevel.MEMORY_AND_DISK)溢写到磁盘。注意cache 不是越多越好只对会被多次复用的 RDD 做持久化否则白白占内存。4. 避坑与排查实验七里最容易翻车的 5 个点4.1 现象任务报 “Python worker failed to connect back”原因PySpark 在本地模式启动 worker 时如果 Python 环境路径不一致或者PYSPARK_PYTHON没设对worker 进程起不来。解决在代码最前面显式指定解释器路径。import os import sys # 强制 driver 和 worker 用同一个 Python 解释器 os.environ[PYSPARK_PYTHON] sys.executable os.environ[PYSPARK_DRIVER_PYTHON] sys.executable参数说明sys.executable是当前运行脚本的解释器绝对路径保证 driver 和 worker 一致。如果你用的是虚拟环境这一步尤其重要否则 worker 会去找系统 Python导致依赖找不到。4.2 现象collect()之后程序卡死或 OOM原因collect把全量数据拉到 driver 内存数据量超过 driver 堆上限就崩。解决用take(n)抽样验证或者把结果saveAsTextFile写回磁盘。# 不要这样数据大时直接 OOM # all_data rdd.collect() # 改成抽样 sample rdd.take(100) # 或者写回存储 rdd.saveAsTextFile(output/result)参数说明take(100)只拉 100 条适合验证逻辑saveAsTextFile的路径不能已存在否则报FileAlreadyExistsException实验前先清理输出目录。4.3 现象reduceByKey结果对但任务数异常多原因分区数设置不合理或者 shuffle 后分区数默认沿用父 RDD 分区数。解决显式指定numPartitions。# 显式控制 shuffle 后的分区数 counts pairs.reduceByKey(lambda a, b: a b, numPartitions4)参数说明numPartitions建议设为集群核数的 2 到 3 倍。设太小并行度不够设太大任务调度开销上升小文件问题也会加重。4.4 现象本地跑通提交到集群报 “ClassNotFoundException”原因本地模式用的是 pip 装的 PySpark集群用的是 Spark 安装包两者版本不一致或者提交时没带 Python 依赖。解决统一版本提交时用--py-files带上依赖包。# 提交任务时指定 Python 依赖和解释器 spark-submit \ --master local[2] \ --py-files deps.zip \ --conf spark.pyspark.python./spark-lab/bin/python \ lab7_job.py参数说明--py-files把额外依赖打包分发到各 executorspark.pyspark.python指定 worker 用的解释器。集群环境下路径要写相对路径或分布式存储路径不能写本地绝对路径。4.5 现象textFile读中文文件出现乱码原因默认按 UTF-8 读如果文件是 GBK 编码就会乱码。解决读进来之后手动转码或者提前把文件转成 UTF-8。# 读进来按字节处理再解码 lines sc.textFile(data/gbk_words.txt).map(lambda s: s.encode(latin-1).decode(gbk))参数说明latin-1是透传编码不改变字节decode(gbk)按实际编码还原。更稳妥的做法是实验前用iconv把文件统一转成 UTF-8避免在代码里做编码转换。5. 从实验到实战把 Spark 初级编程练成可复用的能力实验七做完大多数人就停在“能跑 WordCount”这一步。但如果你想让这次练习真正变成简历上能讲的东西建议做三件事第一把算子按“窄依赖/宽依赖”分类整理成一张表遇到性能问题先判断有没有 shuffle第二用同一份数据分别跑 RDD、DataFrame、SQL 三种写法感受 Catalyst 优化器带来的差异第三把实验里的本地模式代码改造成可传参的脚本用argparse接收输入输出路径这样换数据不用改代码。下面给一个可复用的参数化模板把前面所有练习串起来。import argparse from pyspark.sql import SparkSession def main(input_path, output_path, master): spark SparkSession.builder \ .appName(lab7-reusable) \ .master(master) \ .getOrCreate() # 用 DataFrame API 做同样的 WordCount对比 RDD 写法 df spark.read.text(input_path) from pyspark.sql.functions import split, explode, col words df.select(explode(split(col(value), )).alias(word)) counts words.groupBy(word).count().orderBy(col(count).desc()) # 写回存储coalesce(1) 合并成单文件方便查看 counts.coalesce(1).write.mode(overwrite).csv(output_path) spark.stop() if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--input, requiredTrue) parser.add_argument(--output, requiredTrue) parser.add_argument(--master, defaultlocal[2]) args parser.parse_args() main(args.input, args.output, args.master)逻辑说明SparkSession是 2.x 之后的统一入口DataFrame API 会自动经过 Catalyst 优化执行计划比手写 RDD 更优。coalesce(1)把结果合并成一个文件方便实验检查但生产环境不要这么做会丢失并行度。参数上--master默认local[2]提交集群时改成对应 master 地址即可。验证方法上我一般会做两组对照同一份数据RDD 版和 DataFrame 版各跑一次用 Spark UI 的 4040 端口看 stage 划分和 shuffle 读写量。如果 DataFrame 版的 shuffle 明显更小说明 Catalyst 做了谓词下推或聚合优化。这个对比做一次比看十篇原理文章都管用。最后说个我自己的习惯每次做完实验把当天遇到的报错原文和解决命令记在一个troubleshooting.md里下次遇到直接搜。Spark 的报错信息经常很长但关键就那一两行记多了就有肌肉记忆了。希望帮到你。本文还有配套的精品资源点击获取