简介：围绕Hadoop与Spark框架的数据算法实战源码包，专为大数据开发者、数据工程师以及希望系统学习分布式计算技术的学习者准备。包内共876个文件，包含360个Java源文件、34个Scala源码、242个Jar依赖包、63个Markdown讲解文档、31个Shell运维脚本，以及csv、tsv、parquet等格式的测试数据，压缩包整体约204MB，目录结构清晰，可按章节检索与复用。源码覆盖HDFS文件读写、MapReduce词频统计与数据清洗、RDD算子转换、Spark SQL结构化查询、MLlib分类回归等经典算法场景，并附带可直接运行的输入数据，能直观观察实际处理效果；MapReduce的combiner与partition机制、Spark的shuffle调优等进阶细节也提供了对应示例，非常适合动手实践，有助于深入理解分布式计算的设计思想与调优策略。已有682人浏览学习，是系统提升大数据处理技能、积累真实项目经验的重要参考资料。 我收到过不少读者留言，问题出奇一致：标题里这些词——Hadoop、Spark、数据算法、源代码——单独看都认识，合在一起就不知道从哪里下手。去年有段时间我专门负责带新人做大数据入门训练，发现90%的人卡在同一个地方：不是原理听不懂，而是跟着教程搭完环境之后，跑一个真实的数据算法时，集群就像故意跟你作对一样，各种报错。这篇文章我不打算讲虚的，就沿着一条完整的实战链路，把Hadoop和Spark从关系、搭建、算法代码到性能排查的完整经验拆开，每个环节都附上可以直接抄走改用的内容，希望帮你少走那90%的人走过的弯路。
1. Hadoop和Spark，先分清“存储”和“计算”再动手
1.1 它们不是替代关系，而是搭伙干活
很多新人进大数据这个领域，第一个概念混淆就是：Hadoop和Spark到底学哪个？是不是有了Spark就不需要Hadoop了？
答案是：它们根本不是同一层面的东西。
Hadoop是一个生态体系，核心由三块组成：HDFS负责分布式存储，YARN负责资源调度，MapReduce负责分布式计算。你只要记住一个判断标准——凡是涉及“数据放在哪”的问题，基本都是HDFS的事；凡是涉及“谁来分配算力”的问题，基本是YARN的事；凡是涉及“数据怎么算”的问题，才是计算引擎的事。
而Spark本身只是一位“计算选手”，它不负责存储，也不负责资源管理。它可以把数据从HDFS上读出来，跑完计算再写回HDFS。换句话说，Spark是站在Hadoop肩膀上的计算框架，你的集群可以没有MapReduce，但通常不能没有HDFS。
所以正确的组合方式不是“Hadoop还是Spark”，而是“HDFS + YARN + Spark”。这也是绝大多数生产集群的标准形态。
1.2 数据算法在分布式环境里的真实含义
标题里提到“数据算法”，我理解很多人的第一反应是：是不是要学一堆高深的机器学习公式？
实际不是。在大数据入门和中级应用场景中，所谓的“数据算法”更多是指那些在单机时代很简单、但扔到分布式环境下就需要重新设计的经典算法——词频统计、去重、排序、分组聚合、TopN、Join关联等。
为什么“重新设计”这么重要？给你一个例子：求一个文件里的Top10热词，单机环境下你读文件、建HashMap计数、排序取前10，三分钟完事。但放在分布式环境下，同样的逻辑要考虑：数据分散在多台机器上，每台机器只能算自己那一份；算完局部TopN之后，还需要一个合并动作把所有机器的结果汇总；合并过程如果设计不当，shuffle的数据量可能比你原始数据还大。这就是“设计分布式算法”和“写单机算法”最大的差异。
所以这篇博文后面所有代码都会围绕这个思路展开：先写单机逻辑，再拆成分布式步骤，最后落地成Hadoop或Spark的代码。这也是我认为“数据算法+源代码”最有价值的学习路径。
2. 从伪分布式到真实集群：环境搭建最容易翻车的几个环节
2.1 伪分布式搭建与格式化失败的真正原因
如果你只是想先跑通代码、验证算法，不需要一上来就搞三台机器。Hadoop的伪分布式模式（Pseudo-Distributed）是学习阶段效率最高的方式，一台机器模拟所有角色。
但很多人在这个阶段就会遇到一个极其经典的报错：格式化Namenode时一切正常，启动之后DataNode起不来，日志里报Incompatible clusterIDs。
回想我第一次遇到这个问题的排查过程，也是花了半天才搞明白根因。原因是：你第一次执行hdfs namenode -format之后，NameNode和DataNode各自生成了一套clusterID，它们互相匹配才能正常通信。但如果之后你因为某些操作（比如修改配置）再次执行了格式化，NameNode会生成一套全新的clusterID，而DataNode的data目录里保存的还是旧的clusterID，两边对不上，DataNode自然拒绝注册。
这个问题的解决其实很简单，但网上很多教程没说明白，导致新手反复格式化反复失败：
# 1. 先停掉集群相关进程 stop-dfs.sh # 2. 找到配置文件里指定的数据目录 # 默认在 /tmp/hadoop-${user.name}，或查看 hdfs-site.xml 中的 dfs.namenode.name.dir / dfs.datanode.data.dir # 3. 手动清掉 namenode 和 datanode 的目录 rm -rf /tmp/hadoop-${user.name} # 4. 重新格式化并启动 hdfs namenode -format start-dfs.sh核心思路是：格式化之前，把两边可能残留的旧元数据全部清干净，让NameNode和DataNode在同一次启动中重新生成并协商新的clusterID。
2.2 集群部署策略：学习、面试、生产分别怎么配
再往下一步，你总会面临一个问题：要不要搭集群？搭几台？
我在带新人的时候给过一个经验值，分三种场景：
| 场景 | 机器数量 | 配置建议 | 说明 |
|---|---|---|---|
| 学习/入门 | 1台（伪分布式） | 4核8G起 | 跑通原理和代码足够 |
| 面试/深度练习 | 3台 | 每台4核8G | 1台Master + 2台Worker，能完整演示分布式效果 |
| 生产/真实项目 | 至少5台起 | 每台16核64G以上 | Master节点独立，NameNode和ResourceManager分开 |
很多新手容易犯的错是一上来就租一堆云主机，结果发现集群搭好之后95%的时间都在空转。学习阶段伪分布式完全够用，等到你真正需要调优、测并发、验证数据倾斜时，再上3台集群，这个节奏比较合理。
2.3 高可用必须引入Zookeeper：这里有一个理解门槛
热搜词里有“hadoop和zookeeper整合实战”，这也是一个容易卡住人的主题。
当你有3台以上机器时，NameNode会成为整个HDFS的“单点”——一旦它挂了，整个集群就无法读取文件元数据，等于全部瘫痪。解决方式就是高可用（HA）方案：跑两个NameNode，一个Active，一个Standby。
Zookeeper在这里的角色是什么？简单说，它是一个“分布式协调器”。当Active NameNode出现故障时，Zookeeper需要快速感知到这个故障，然后触发自动切换，把Standby NameNode提升为Active。整个过程不需要人工干预。
配置HA时的核心文件是hdfs-site.xml，你需要在里面声明nameservices、两个NameNode的地址、以及Zookeeper的仲裁地址。这里强调一个最容易踩的坑：dfs.namenode.shared.edits.dir必须指向一个JournalNode集群，很多人在这一步图省事直接复用DataNode节点，导致故障切换时元数据无法同步，HA形同虚设。
3. 数据算法落地：从MapReduce到Spark的源代码拆解
3.1 MapReduce版WordCount：理解分布式计算的“最小单元”
WordCount对大数据的意义，相当于HelloWorld对编程的意义。不要因为它简单就跳过，它实际上是理解分布式计算模型的最好载体。
一个完整的MapReduce任务包含三个核心组件：Mapper、Reducer、以及驱动类。Mapper负责把数据切成KV对，Reducer负责对相同Key的Value做合并。
public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }这段Mapper代码的逻辑很简单：读进来一行文本，按空格切分成单词，每遇到一个单词就输出一个(word, 1)的键值对。
public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }Reducer做的事情更简单：把所有相同单词的计数加起来，得到一个最终计数值。这里的关键在于，Map阶段产生的中间数据会经过一个叫Shuffle的过程，由框架自动把相同Key的数据路由到同一个Reducer上，你在代码里完全不用管这个逻辑——这也是MapReduce被称为“隐藏了分布式复杂性的模型”的原因。
驱动类不再完整贴出，核心就一行：
job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class);3.2 用Spark RDD重写同一算法：对比着看才看得出差距
等你理解了MapReduce的套路，再用Spark重写同一个算法，你会立刻感受到“代码量”和“思维复杂度”的双重下降。同样一个WordCount，Spark写法如下：
val lines = sc.textFile("hdfs:///input/data.txt") val wordCounts = lines .flatMap(_.split(" ")) .map(word => (word, 1)) .reduceByKey(_ + _) wordCounts.saveAsTextFile("hdfs:///output/wc_result")如果你已经看懂了MapReduce的版本，这段代码你甚至不需要注释就能读懂：flatMap负责把每行拆成单词，map负责把单词变成KV对，reduceByKey负责按Key聚合计数。
但你注意到没有，这中间少了一个非常重要的东西：Shuffle。MapReduce里，每个单词和它的计数从Mapper流向Reducer的过程，是需要落盘（写到磁盘）的；而在Spark里，reduceByKey会先在每个分区内部做一个局部聚合，再把合并后的结果通过网络传输到下游节点，大幅减少了需要搬运的数据量。这也是Spark“比MapReduce快”的核心原因之一，不只是因为它在内存里算，更因为它减少了很多不必要的磁盘和网络开销。
3.3 字数统计之外：TopN和分组排序的两种写法
如果你真正想应付面试和实际需求，WordCount之后建议立刻啃掉TopN问题。它在分布式环境里比WordCount多了一层考验：如何避免把所有数据都shuffle到一个节点上。
先说最简单但不推荐的做法：
// 全局排序取TopN：数据量大时很容易造成OOM val topN = wordCounts.sortBy(_._2, ascending = false).take(10)这种方法写起来很爽，但问题是sortBy通常是全局排序，需要把所有数据集中处理，数据量一旦上来，性能急剧下降。
更推荐的做法是“两步走”思路：先在每个分区内部取局部TopN，再对局部TopN结果合并取全局TopN。
// 第一步：每个分区内部统计Top10 val localTop = wordCounts.mapPartitions { iter => iter.toList.sortBy(-_._2).take(10).iterator } // 第二步：对局部结果做全局Top10 val globalTop = localTop.sortBy(_._2, ascending = false).take(10)这种写法的精髓在于：每个分区只需要输出10条数据，最终需要shuffle的数据量被压到了可以忽略不计的量级，整个作业性能会有一个质的提升。
3.4 结构化数据直接用Spark SQL，别自己造轮子
如果你要处理的数据本身有结构（JSON、Parquet、CSV带表头），我建议直接上Spark SQL，而不是自己写RDD逻辑。代码可读性和维护性是RDD写法完全没法比的。
val df = spark.read.json("hdfs:///data/events.json") df.createOrReplaceTempView("events") val result = spark.sql( """ |SELECT user_id, COUNT(*) AS cnt |FROM events |WHERE event_type = 'click' |GROUP BY user_id |ORDER BY cnt DESC |LIMIT 10 |""".stripMargin) result.show()这段SQL如果你有一点关系型数据库的基础，几乎零学习成本。Spark SQL底层的Catalyst优化器会自动做谓词下推、列裁剪等优化，很多时候比你手工写RDD性能还好。这就是“选择比努力重要”的典型场景——同样的结果，你用RDD硬写可能还需要调优半天，换Spark SQL一行优化都不用做。
4. 性能疑案：Spark on YARN明明配了4核，为什么只用了1个？
4.1 从一次真实排查看CPU配置的完整链路
热搜词里有一条让我印象很深：“spark on yarn cpu只能用1个是为什么”。这个问题我印象深，是因为当初带项目时真被它坑过，而且网上答案非常分散。
当时的情况是这样：用户用spark-submit提交作业时，明明通过--executor-cores 4指定了每个Executor要4个CPU核，但在YARN的资源管理界面上看到的实际使用核数始终只有1个，作业跑得奇慢无比。
排查过程我按这个链路走：
第一步，确认spark-submit的参数是否生效：
spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-cores 4 \ --executor-memory 8g \ --num-executors 3 \ your-job.jar参数看起来完全没毛病。
第二步，去看YARN的调度配置。如果你的YARN被配置成了yarn.scheduler.capacity.resource-calculator或者DominantResourceCalculator，在分配资源时是按“CPU + 内存”整体来算的。如果每个容器的yarn.nodemanager.resource.cpu-vcores没有正确设置，YARN根本不知道每台机器有几个CPU可以用，所有容器都会被默认限制为最小值1个vCore。
第三步，去检查yarn-site.xml里这两个参数：
<property> <name>yarn.nodemanager.resource.cpu-vcores</name> <value>8</value> </property> <property> <name>yarn.scheduler.maximum-allocation-vcores</name> <value>8</value> </property>问题往往就出在这里：cpu-vcores没有显式设置成物理机的真实核数，YARN默认只认1个核，然后无论你--executor-cores写几，都会被限制成1。把这两个参数按实际核数配置后，重启YARN集群，再提交同样的任务，CPU核数就正常了。
4.2 Spark内存模型：为什么executor-heap设了10G还是OOM
提到CPU就绕不开内存。很多新人设置--executor-memory 10g，以为Executor最大就能用10G，结果跑着跑着频繁报OOM，非常困惑。
这里的关键在于Spark对Executor内存做过精细划分。在Executor的堆内内存里，主要分为三块：
- Execution内存：用于shuffle、join、sort等计算过程
- Storage内存：用于缓存RDD数据
- Reserved内存：留给系统自身使用（默认300MB）
spark.executor.memory设置的是Executor进程的总堆内存，但这部分内存并不是“纯计算可用内存”。实际执行计算时，能够真正被一个task使用的内存，需要按spark.memory.fraction（默认0.6）再乘以spark.memory.storageFraction（默认0.5）来计算。
这是很多人OOM的真正原因：你设置了10G堆内存，但减去Storage保留的30%左右，真正的Execution内存可能只有4G左右。如果作业里Join或Aggregate操作的数据量远超这个值，很容易出现内存溢出，而它并不是因为你“总内存不够”。
调优的方向有两类：一是增加Executor内存；二是降低spark.memory.storageFraction，把更多内存让给Execution。
4.3 一套可以直接参考的配置基线
根据不同作业类型，我自己常用的配置从这样一个表起步：
| 参数 | 建议值 | 适用场景 |
|---|---|---|
| --executor-memory | 8g-16g | 通用默认 |
| --executor-cores | 4-8 | 建议不超过物理机核数的1/4 |
| --num-executors | 依据队列资源 | 单作业不建议超过50 |
| spark.memory.fraction | 0.6-0.75 | 常规即可 |
| spark.sql.shuffle.partitions | 200 | 集群规模较大时可调大 |
这个表不是标准答案，但作为初始值基本不会出大问题。后续再结合监控面板里Execution内存的波峰波谷做精细调整。
5. 大数据面试高频考点：宽窄依赖、数据倾斜、Spark快在哪
5.1 Spark为什么快？别再回答“因为有内存”
面试官特别喜欢问“Spark为什么比MapReduce快”，但绝大多数人只会回答“因为Spark基于内存计算”。这个答案没有错，但是远远不够。
专业一点的回答应该包含三个层次。第一层，内存计算确实减少了大量磁盘I/O，尤其是迭代式算法；第二层，Spark引入的DAG调度引擎，能够记住整个计算流程的依赖关系，属于“懒执行”模式，前面有多个操作时能自动合并优化；第三层，也就是我在前面提到过的，Spark在很多操作上会做分区内聚合，减少Shuffle数据量。
你如果能从这三层来回答，面试官会立刻知道你是真正写过代码做过调优的，而不是背了八股文。
5.2 RDD的宽窄依赖：理解数据倾斜的理论基础
另一个高频考点是“宽依赖和窄依赖的区别”。我习惯用生活化方式解释给新人听。
窄依赖就像每个家长只负责自己的孩子，父RDD的每个分区最多被子RDD的一个分区使用，不需要跨节点传输，代表操作有map、filter。宽依赖就像开家长会，所有家长（父分区）都要把孩子（数据）送到同一个班级（子分区），代表操作有groupByKey、reduceByKey、join。
这个区别如此重要，是因为宽依赖会触发Shuffle，而Shuffle是整个作业性能瓶颈的第一来源。数据倾斜（某个Key的数据量远远超过其他Key）之所以那么难搞，本质也是因为Shuffle阶段把大量数据集中到了同一个节点上，导致部分节点超时、部分节点空闲。
针对数据倾斜，一个最实用的处理思路就是“把倾斜的Key单独拆出来处理”。先把大数据量的Key过滤出来单独跑一个作业，其余的Key正常跑，最后合并两段结果。这个思路简单粗暴，但解决90%的倾斜问题都够用了。
5.3 学习路线建议：先把一条链路走通再横向铺开
最后聊一点路线问题。我自己比较推荐的学习节奏是：先花一周时间把MapReduce WordCount完整走通，理解Mapper、Reducer和Shuffle的基本流程；再用一周时间把同样的算法用Spark RDD重新实现一遍，对比两者的差异；接着用Spark SQL处理一个真实数据集，把数据清洗、聚合、TopN这些基础操作跑熟；最后再回头去补理论，比如宽窄依赖、内存模型、调度原理。
这个路线的特点，是每个阶段都能产出看得见的结果。每天都能有一小段可运行的代码“跑通”，这种正反馈对学习来说是至关重要的。很多人的问题不是不努力，是花大量时间在“看教程”而不是“写代码”，导致原理讲得头头是道，一打开终端就卡在环境变量上。数据算法这个东西，动手永远比看教程重要十倍。
像“源代码”这块，我的建议是不要满足于把示例代码复制粘贴跑出结果。你可以做这样一个实验：把WordCount的Reducer逻辑从“求和”改成“求平均数”，或者把TopN从“取最大”改成“取最小”，每次改一个点，看看输出变化。改动和输出之间的因果链路建立起来之后，你对分布式算法的理解深度，会远超那些只刷过十遍教程的人。
本文还有配套的精品资源，点击获取
