Spark入门从RDD开始：核心机制与常用算子实操路线

前阵子有个刚转大数据方向的后辈问我现在Spark上都流行写SQL和DataFrame还有必要去啃RDD编程吗我说有必要而且强烈建议先把RDD这个底子打扎实。RDD是Spark最核心的数据抽象也是理解整个Spark执行框架的钥匙绕开它直接学SQL你始终是隔着一层皮在写代码。这篇实验记录就是给没接触过Spark的人准备的入门路线从环境搭建、第一个RDD程序到常用算子的完整实操和踩坑排查全程都用本地模式跑不需要集群一台普通电脑就够。适合读者很明确刚接触Spark的学生、从Java或Python后端转大数据的开发以及已经会写Spark SQL但始终搞不清底层是怎么回事的中间层工程师。看完这篇你可以独立跑通一个RDD实验并且知道每个算子在分布式环境下到底发生了什么。1. 为什么现在学Spark我仍建议先从RDD动手1.1 入门者的常见误区先学SQL真的更快吗很多教程为了追求快速上手一上来就教Spark SQL和DataFrame的API效果确实立竿见影读个文件、跑个SQL几分钟就有结果。但我会建议初级实验老老实实从RDD开始原因很直接SQL是结果RDD才是过程。举个不太严谨但很贴切的类比DataFrame相当于给你一辆已经调好的自动挡汽车你踩油门就走RDD则要求你先打开引擎盖搞清楚发动机、变速箱、传动轴大概是怎么配合的。日常开车当然不需要打开引擎盖但在Spark里跑任务一旦性能出问题、数据倾斜、OOM你只有理解底层RDD的执行模型才知道应该朝哪个方向调整。我在带新人做实验时观察到一个现象直接学DataFrame的人调参全靠猜试试增加分区数、换个写SQL的方式不行就再试而先做一遍RDD实验的人会习惯性地思考这一步是不是触发了shuffle这个算子是不是要全量落盘排查问题的思路完全不一样。1.2 RDD到底是什么一个带着分区的弹性数据集合RDD全称Resilient Distributed Dataset中文常译作弹性分布式数据集。拆开看三个词就抓住了本质Resilient弹性的指容错能力某个分区的数据计算失败Spark可以根据血缘关系重新计算不需要整份数据重来。Distributed分布式的数据天然被切分成多个分区分布在不同Executor上可以并行处理。Dataset数据集本质上是一个分布式的数据集合你可以像操作本地Collection一样对它做map、filter等操作。但RDD有一个初学者需要立刻建立的概念它不保存真正的数据它保存的是如何从数据源计算出这些数据的描述。这个概念在后续理解惰性执行时非常关键。你可以把RDD理解成一份菜谱数据是食材执行Action时才会真正起火做菜。1.3 学RDD的长期收益执行机制和排错能力的地基做这次初级实验你的直接输出是一个能跑的词频统计、日志清洗程序这是短期收益。长期收益在于三个层面一是你真正理解Spark程序怎么写才高效。知道某个操作是窄依赖还是宽依赖就能判断什么场景会卡在网络传输上知道惰性执行就不会在一个循环里去反复触发Action。二是调问题有方向。日志报错、数据倾斜、shuffle异常这些高级问题的表象千奇百怪但根因都落在分区、依赖、缓存机制上。RDD实验正是认识这些机制的最低成本路径。三是面试和后续学习都绕不开它。很多分布式框架都借鉴了RDD的思路理解RDD对学Flink等其它引擎也有迁移价值。2. 动手之前RDD的三大核心机制决定你该怎么想问题2.1 分区数据是怎么被拆开并行处理的RDD里的数据会被切分成若干个分区Partition分区是Spark并行计算的基本单位。一个分区对应一个任务TaskExecutor上的一个核同一时间处理一个任务。所以分区数越多、并行度越高这个说法不算错但不全面分区数也不是越多越好。分区数怎么定创建RDD时可以指定不指定时也有默认规则。我用一个例子说明from pyspark import SparkContext, SparkConf conf SparkConf().setAppName(partition_demo).setMaster(local[*]) sc SparkContext(confconf) # 将本地集合转换成RDD显式指定3个分区 data [1, 2, 3, 4, 5, 6, 7, 8] rdd sc.parallelize(data, 3) print(分区数量, rdd.getNumPartitions()) print(每个分区的数据, rdd.glom().collect())glom()是个很适合实验阶段观察数据分布的算子它把每个分区的内容收集成一个数组返回一个列表。上面这段代码的输出会让你对数据被切分有直观印象本地集合被拆成了3份每份数据数量并不要求均匀。用textFile读取文件时分区数通常由文件大小和HDFS块大小决定本地模式下会按输入格式自动估计。不过初级实验更建议显式给定分区数不确定设置多少时我常用的取值是local[*]加默认配置跑通功能后再回去调。为什么分区不能太大也不能太小分区太少集群空闲跑得慢分区太多任务调度开销和网络传输又可能让总体时间变长。这个尺度感不是看几篇文章就能建立的需要你在实验里故意设置成1、2、4、8分别跑一次观察执行时间、观察Spark UI里的任务数量经验很快就会积累起来。2.2 惰性执行Transformation和Action的关键区别RDD算子分成两大类Transformation转换算子和Action行动算子。我先不说定义直接看代码# 这一步只是记录了一个待执行的指令 rdd2 rdd.map(lambda x: x * 2) print(Map执行了吗) # 这一步才真正触发了计算 total rdd2.sum() print(总数, total)运行上面代码你会发现在print(Map执行了吗)的时候Spark日志里没有任何执行信息因为map这个操作只是构建了一个新的RDD描述没有真正读取数据。直到sum()这个Action出现Spark才开始从前到后执行整条计算链。为什么Spark要设计成惰性的因为延迟执行能给引擎提供整体优化空间。比如你有连续的map、filter操作Spark可以把它们合并到同一个阶段Stage里减少中间结果的落盘和传输。这种优化思路有点像编译器的指令重排追求的是整体最优而不是每条语句立即生效。从学习角度你只需要记住Transformation生成新RDD不计算Action返回结果或落盘才真正计算。常见Transformation有map、flatMap、filter、groupByKey、reduceByKey常见Action有collect、count、take、first、foreach、saveAsTextFile。2.3 血统机制为什么分布式计算失败后还能重算RDD之间会记录我是谁生出来的这种关系这个记录叫血缘Lineage或血统。比如rdd3 rdd2.filter(...)那么rdd3知道自己的父RDD是rdd2rdd2知道自己的父RDD是rdd1一直可以追溯回数据源。这个设计解决了分布式系统最头疼的容错问题。你运行一个100个分区的大任务中间第40个分区所在的节点挂了数据丢了怎么办Spark通过血统找到这个分区对应的父RDD重新读取数据源或重新执行之前的转换就能把丢失的分区算回来而不需要整个任务重跑。这也是它取名弹性的重要原因。在初学阶段血统对你最直接的意义是当你看到执行失败的错误栈顺着血缘关系一层层往上查通常能定位到是哪个算子、哪一步输入出了问题。Spark UI里的DAG图画的就是这条血缘链路。做实验时不要只盯着跑没跑通点开作业页看一眼DAG图每一个框对应哪一步转换这是白捡的实战经验。3. 实验环境准备本地模式跑通第一个RDD程序3.1 语言选择Python还是Scala更适合初级实验Spark支持Java、Scala、Python、R四种语言。如果目标是快速理解RDD编程模型我推荐Python。原因不只是代码简洁更重要的是Python的交互式体验好可以在pyspark或者JupyterNotebook里逐行执行、逐个查看结果这个反馈速度对新手非常友好。Scala有性能优势和类型安全是Spark的原生语言很多生产代码用Scala写相关面试题也爱围绕Scala版本提问。但如果你的职业方向不是专职Scala开发完全不必在入门实验阶段硬啃。Java则更容易被冗长的样板代码干扰学习主线我不建议初级实验选Java。我这次实验记录全部用PySpark风格展示你只需要会基础Python语法lambda表达式、列表、字典就能顺畅读下来。如果你将来要做深度二次开发或阅读Spark源码再补Scala不迟。3.2 安装环境与版本搭配一次装对少踩半小时坑本地模式不需要安装HDFS和集群只要装一套Spark发布包和对应语言环境即可。我建议你在实验前把三样东西准备好JDK 8或JDK 11注意和Spark版本的兼容关系Spark 3.4及以上官方推荐JDK 8/11/17Python 3.8及以上如果用PySparkSpark发布包注意区分版本号下载时选择不带Hadoop的版本在本地模式跑反而更省事因为本地模式会直接读你硬盘上的文件。Windows上我提醒几个点Spark的bin目录里有些脚本依赖winutils.exe如果直接跑spark-shell或pyspark报错提示找不到Hadoop相关文件把HADOOP_HOME配好并放入对应版本的工具文件通常就能解决。macOS和Linux则基本没有这个问题。启动方式有两种一种是交互式# 进入PySpark交互环境 pyspark另一种是把脚本保存成.py文件用spark-submit提交spark-submit rdd_lab.py3.3 创建RDD的三种方式parallelize、textFile与DataFrame转换实验阶段只要掌握三种创建方式后面会反复用到。第一种从本地集合创建rdd1 sc.parallelize([apple, banana, orange, apple], 2)sc.parallelize适合生成小规模测试数据参数可以选择分区数。你没有外部文件但又想验证某个算子时这是最快的方式。第二种从外部存储读取rdd2 sc.textFile(data/access_log.txt)textFile按行读取文件每一行就是RDD中的一个元素。这个方式最接近真实场景生产环境读HDFS等分布式存储也是同一套API。注意这里对路径的处理本地模式有点击文件路径集群模式下要换成存储路径初学者容易在这里混淆。第三种从DataFrame转换df spark.read.csv(data/orders.csv, headerTrue) rdd3 df.rddDataFrame转化来的RDD每个元素是一个Row对象访问字段时不建议用前面的tuple解构方式更容易出错的场景会在后面文章里讲到。简而言之同一份数据你走得通文件到DataFrame再到RDD的链路就掌握了Spark两种执行引擎之间的桥梁。图省事的学生往往只练第一种结果一进真实项目从textFile开始就卡住。三种方式我都建议在实验里各建一个看看分区数和内容。3.4 第一次查看数据collect、take、foreach的取舍RDD造出来之后怎么确认里面的数据对不对新手第一反应常常是collect()它能返回全部数据。而我会告诉你实验数据量小collect随便用一旦数据量大了collect()会把所有数据拉到Driver端内存直接溢出。所以更稳妥的探索方式是take(n)只取前n个元素返回给Driverfirst()取第一个foreach(print)在Executor端打印适合少量日志输出glom().collect()查看分区分布。看下面这个综合的第一次实验脚本from pyspark import SparkConf, SparkContext conf SparkConf().setAppName(first_rdd_lab).setMaster(local[*]) sc SparkContext(confconf) sc.setLogLevel(WARN) rdd sc.textFile(data/access_log.txt) print(总行数, rdd.count()) print(前3行, rdd.take(3)) rdd.take(5).foreach(print) sc.stop()sc.stop()建议在脚本末尾养成习惯它释放集群资源。交互环境里不用每次执行这个但提交脚本一定要有否则Driver可能会一直占用资源。4. 算子实操从词频统计到订单聚合的完整演练4.1 词频统计理解flatMap-map-reduceByKey链路词频统计在RDD实验中的地位约等于编程界的Hello World。它麻雀虽小却包含了RDD编程最重要的三个算子。假设你要统计某个文本文件里每个单词出现多少次标准写法是lines sc.textFile(data/words.txt) words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) result counts.collect() for word, num in result: print(word, num)逐个拆解flatMap一行文本拆成多个词。它和map的区别在于map的输入与输出是一对一flatMap则是一对多输出会被展平成一个个单独元素。map(lambda word: (word, 1))每个单词转成一个键值对这是MapReduce经典思想中的标记步骤。reduceByKey(lambda a, b: a b)相同key的value合并相加。reduceByKey是这次实验要重点体会的一个算子。它先在每个分区内部对相同的word做局部求和再把局部结果汇聚到最终分区做全局求和。这种先局部后全局的操作大大减少了跨节点传输的数据量。你可以故意把代码改成先用groupByKey再手动sum对比一下执行计划中shuffle的数据量或者看Spark UI中不同任务的耗时。很多概念讲一百遍不如你亲手对比一次。4.2 数据清洗filter、distinct、union的实际用法词频统计跑通后我把实验场景升级一下处理一份模拟访问日志。文件每行是click,page_home这种格式第一段是行为类型第二段是页面名。你要过滤出只保留click行为的记录同时去掉重复页面名。# 读取日志过滤行为类型为click的行 raw sc.textFile(data/access_log.txt) clicks raw.filter(lambda line: click in line) # 提取页面名 pages clicks.map(lambda line: line.split(,)[1]) # 去重 unique_pages pages.distinct() print(click行为数, clicks.count()) print(去重页面数, unique_pages.count())filter是最直观的转换算子但要注意它返回新的RDD原RDD不会变这是函数式编程的惯例。初学阶段很多人误以为原数据被过滤了画一下数据流图马上就不会混淆。distinct使用起来很简单但背后会触发shuffle因为想要全局去重必须确认某个值在所有分区中都只出现一次本质上是聚合操作。在实验里你可以对比count()和distinct().count()的耗时差距感受一下shuffle的成本。union则用于合并两个RDD比如合并昨天的日志和今天的日志today_log sc.textFile(data/log_today.txt) yesterday_log sc.textFile(data/log_yesterday.txt) all_log today_log.union(yesterday_log)注意union不保证去重它不是集合语义的并集更准确说是拼接。需要去重时接着加一个distinct。4.3 聚合与排序reduceByKey和groupByKey的对比做日志分析除了统计行为数最常遇到的就是按维度分组后聚合。假设你有一份模拟订单数据格式是订单ID,商品类别,订单金额要统计每个类别的订单总金额和平均金额。orders sc.textFile(data/orders.txt) # 解析成 (类别, 金额) pairs orders.map(lambda line: line.split(,)).map(lambda arr: (arr[1], float(arr[2]))) # 总金额与单数 total_amount pairs.reduceByKey(lambda a, b: a b) order_count pairs.mapValues(lambda v: 1).reduceByKey(lambda a, b: a b)这里出现了一个初学者很想问的问题reduceByKey和groupByKey到底应该用哪个我的建议是默认先想reduceByKey。原因在于groupByKey会先把相同key的所有value全部收集起来形成(key, 可迭代集合)这是全量shuffle而reduceByKey在每个分区先做局部聚合把多个元素压缩成一个局部结果再参与全局shuffle传输的数据量小得多。两者对比用表格看得更清楚算子局部聚合触发的shuffle适用场景reduceByKey有数据量被压缩后再传输求和、计数、最大值最小值groupByKey无同一key的value全量传输需要拿到完整分组集合做逐元素处理处理完要按金额排序输出简单方案是对结果调用sortBy它是Action前的最后一步result total_amount.collect() sorted_result sorted(result, keylambda item: item[1], reverseTrue) print(sorted_result[:10])也可以直接用RDD提供的sortByKeysorted_rdd total_amount.sortByKey(ascendingFalse)排序同样有shuffle做实验时观察一下执行计划里多出来的Stage就明白了。4.4 结果落盘saveAsTextFile与分区数配合实验的最后一步通常要把结果保存下来。saveAsTextFile会把结果输出到文件而且写出来的是前面提到的分区数量决定的多个文件不是单独一个大文件counts.saveAsTextFile(output/word_count_result)新手第一次看到输出目录里出现一堆.part-00000、.part-00001文件会不习惯这是正常现象因为每个分区的计算结果各自写入一个文件。如果嫌小文件太多可以用coalesce(1)先合并分区再输出counts.coalesce(1).saveAsTextFile(output/word_count_single)coalesce(1)会把数据合并成一个分区减少文件数但这会导致单任务处理全部数据失去并行度。生产环境里通常不会合并成1个而是调整到合理分区数。这里引出一个更重要的话题分区数怎么设置我放在后面的踩坑章节详细讲。5. 实验中最容易翻车的三个问题与排查思路5.1 只写Transformation不写Action程序无反应第一次做实验最常见的疑问是我明明写了正确的map和filter为什么运行半天什么结果都没打印这时候判断逻辑很简单——检查自己有没有调用Action。我见过一个同学是这样写的rdd sc.textFile(data/access_log.txt) rdd.map(lambda line: line.upper()) print(处理完成)他期望打印出来的日志内容是小写转大写实际只打印了处理完成。因为map只是定义了转换Spark并不知道你要数据有什么用自然不执行。修复方式很简单加一个collect()或者take()rdd sc.textFile(data/access_log.txt) converted rdd.map(lambda line: line.upper()) print(converted.take(3))这个问题的排查思路很标准化从头看你的算子清单最后一个算子是不是Action不是就补上。熟悉之后你会主动把多个Transformation串成一条链最后用一个Action收尾既清晰又高效。5.2 分区数设置不当并行度过高让小文件变灾难第二个高频问题出现在保存结果时。有人想加快处理速度一开始就把分区数设很大比如parallelize(data, 200)最终结果落盘产生200个小文件打开一看每个文件里只有几行数据。这在小规模实验里看着别扭在生产环境可能是灾难文件数和后端元数据压力都会直线上升。问题根源分区数是并行度和输出文件数之间的杠杆。并行度太低跑得慢太高每个分区的数据量太少任务调度开销超过计算收益输出文件又多。实验阶段该怎么做我建议按经验值设置本地实验数据量在几MB级别分区数设置为2到8就够需要查看结果时可以用coalesce合并到1个分区输出注意coalesce(1)比repartition(1)更节省shuffle成本因为它只减少分区不增加如果想均匀调整分区数用repartition(n)它会触发shuffle重新分配数据。# 减少分区 rdd_low_partition rdd.coalesce(1) # 增加分区会触发shuffle rdd_high_partition rdd.repartition(8)踩几次坑后你会形成条件反射处理数据之前想一遍分区数保存结果之前再想一遍文件数。5.3 日志刷屏看不到关键信息调整级别与使用本地UI第三个问题不是程序跑挂了而是日志刷得太厉害把结果淹没了。启动Spark后控制台会打印大量INFO级别日志如果你的实验结果只有一行很容易被冲掉。排查的办法分两步。第一步把日志级别调成WARNsc.setLogLevel(WARN)这句代码放在程序开头它告诉Spark把INFO级别的日志过滤掉只保留警告和错误。这样print的输出就很显眼。第二步打开本地WebUI观察任务。local模式下Spark会在本地启动一个监控页面默认从4040端口开始自动递增。任务运行期间打开它可以看到当前运行了多少Stage、多少任务、每个任务的耗时、数据输入量分布。这是初学者认识Spark运行结构的最好窗口。运行结束后页面会关掉所以要看就得趁程序跑着的时候看或者在交互环境里持续打开。看UI的时候重点看两块一是DAG图理解数据流二是Shuffle Read Size那些数据量大到异常的Stage潜力问题往往在那里。6. 初级实践之后缓存、广播变量和下一个学习方向6.1 缓存和复用persist与cache的正确姿势词频统计里如果你的原始日志clicks要被多个Action复用每次Action都会从头读取文件再重算整个链条数据一大会非常浪费。这时就该引入缓存。from pyspark import StorageLevel clicks.persist(StorageLevel.MEMORY_AND_DISK) print(clicks.count()) print(clicks.filter(lambda line: home in line).count())cache()是persist(StorageLevel.MEMORY_ONLY)的简写。它们的共同点是把中间结果跨过Action保存下来下一次Action直接读取缓存不再重算。我建议初级实验里至少实验一次persist。你可以故意把某个RDD连续执行多个Action对比使用缓存前后的执行时间再观察Spark UI中同一个Job两次执行的耗时差异比任何文案都直观。需要注意缓存是懒执行的真正缓存发生在第一次Action之后。另外缓存的数据量大且超出内存时MEMORY_ONLY会反复重新计算改用MEMORY_AND_DISK是更稳妥的选择。6.2 广播变量与累加器跨节点传数据的两种手段RDD编程到了第二个进阶点是跨节点的数据传递问题。你在Driver端定义了一个字典想在算子里用每个Executer都会拷贝一份完整副本数据量大时通信开销很大。解决方案是用广播变量# 模拟一个省份映射表 province_map {BJ: 北京, SH: 上海, GZ: 广州} bc_map sc.broadcast(province_map) rdd sc.parallelize([(BJ, 100), (SH, 200), (BJ, 80)]) result rdd.map(lambda t: (bc_map.value.get(t[0]), t[1])).collect() print(result)广播变量的核心思想是大对象只分发一次所有任务共享读取而不是每个任务传递一份序列化副本。和它配套的是累加器常用于Driver端汇总统计、计数error_count sc.accumulator(0) rdd sc.textFile(data/access_log.txt) def check_error(line): if error in line: error_count.add(1) rdd.foreach(check_error) print(错误行数, error_count.value)执行完Action之后error_count.value是累计值。初学阶段广播变量和累加器作为两个补充工具认识即可但它们的适用场景会在实际项目中反复出现。6.3 从RDD到DataFrame什么时候该切换什么时候该回头做完整套初级实验你应该能感受到RDD API的优雅与繁琐并存。它在复杂、非结构化数据的处理上有极大灵活性不需要预定义Schema但它写起来偏底层优化的工作要自己做不少。DataFrame则在RDD之上加了一层结构化信息SchemaSpark可以利用这些信息做更智能的优化比如列裁剪、谓词下推。同样一份数据用DataFrame API经常比手写RDD算子快不少代码也更短。这就是官方越来越推荐DataFrame的原因。但什么时候该回头用RDD我的经验是数据源是非结构化的文本处理逻辑依赖复杂自定义函数或者你在调试一个不确定性能瓶颈在哪的问题时RDD的透明度和可控性反而是优势。你同时掌握两条路才算真正具备Spark的开发能力。6.4 给初级实践者的一条串联式练习建议不要在跑完几个算子后就算完成任务。你只需要准备一个模拟日志文件把这次实验所有的算子串成一条完整链路它的步骤大致是用textFile读取原始日志用filter过滤出有效记录用map或flatMap做字段解析用reduceByKey统计分组指标用sortBy输出TopN用saveAsTextFile保存结果到一个合理分区数的目录。我在做这次初级实验时最大的感受是分布式编程的很多坑不是那种看一眼定义就能避开的坑而是只有亲手把数据放进去跑一遍、再亲眼看到它输出不对的时候才算真正长在脑子里的经验。这篇内容最值得你带走的其实不是某一个算子的用法而是遇到问题时的排查顺序先确认有没有Action再看分区数是否合理最后用UI确认shuffle和耗时。走完这一轮你的RDD实验就不只是交作业而是真正入门了。