Hadoop-Spark-大数据双雄从入门到真香

Hadoop/Spark大数据双雄从入门到真香假设你是一家电商公司的后端开发某天老板丢过来一个需求“把咱们过去三年的订单数据分析一下看看用户购买行为有啥规律。”你看了看数据库里那张 10 亿行的订单表陷入了沉思…一、问题来了数据太大单机扛不住咱们先从最朴素的思路开始。你想这不就是写个 SQL 做聚合嘛简单SELECTuser_id,COUNT(*)asorder_count,SUM(amount)astotal_amountFROMordersWHEREcreated_at2022-01-01GROUPBYuser_id结果一跑MySQL 直接罢工了——查询跑了半小时还没出结果CPU 飙到 100%其他业务查询也跟着卡死。问题来了数据量太大TB 级别单台机器内存根本装不下计算太慢CPU 单核性能再强也架不住海量数据一挂全挂没有容错中途崩了就得重来说白了单机已经到天花板了得想办法让多台机器一起干活。这时候Hadoop 和 Spark 就登场了。二、Hadoop大数据的开山鼻祖2.1 核心思想分而治之Hadoop 的核心理念特别简单——既然一台机器搞不定那就把数据切分成小块扔到多台机器上并行处理。就像你有一堆砖要搬一个人搬太慢那就叫上一群人每人搬一摞搬完再汇总。Hadoop 主要包含两大组件组件作用类比HDFS分布式文件存储一个超大号的网盘数据分散存在多台机器上MapReduce分布式计算框架把大任务拆成小任务分发到各节点执行2.2 HDFS数据怎么存HDFS 的设计很巧妙# 伪代码HDFS 文件存储原理 class HDFS: def store_file(file): # 1. 把大文件切成 128MB 的小块Block blocks split(file, block_size128MB) for block in blocks: # 2. 每个块存 3 份容错 replicas [node1, node2, node3] for node in replicas: node.write(block) # 3. NameNode 记录哪个块存在哪台机器 name_node.record_metadata(file, blocks, replicas)关键点切分存储大文件切成 128MB 的块分散存到不同机器三副本机制每个块存 3 份挂了一台机器数据不会丢NameNode 管目录记录文件的元数据存在哪、有多少块DataNode 管存储真正存数据的打工节点2.3 MapReduce计算怎么做MapReduce 是 Hadoop 的计算引擎编程模型就两个阶段# 伪代码MapReduce 计算单词频次# Stage 1: Map映射# 每台机器并行处理自己本地的数据块defmap(line):wordsline.split()forwordinwords:# 输出键值对(word, 1)emit(word,1)# Stage 2: Reduce归约# 把相同 key 的数据汇总到一起计算defreduce(word,values):# values [1, 1, 1, 1, ...]totalsum(values)emit(word,total)执行流程输入文件 → [Split] → Map 并行处理 → [Shuffle] → Reduce 汇总 → 输出结果 ↑_________________↑ 这一步自动帮你做 把相同 key 的数据拉到同一台机器MapReduce 的优缺点优点缺点高容错挂了几台节点不影响中间结果要写磁盘太慢扩展性强加机器就能提速编程模型太死板复杂逻辑写起来很绕适合批处理吞吐量大不适合迭代计算比如机器学习说白了Hadoop MapReduce 就像一台老式的柴油拖拉机——力气大、皮实耐造但是启动慢、油耗高跑起来轰隆隆的。这时候Spark 出现了。三、Spark大数据计算的涡轮增压3.1 为什么 Spark 更快Spark 和 Hadoop 最大的区别就一句话Spark 尽量把数据放在内存里算Hadoop MapReduce 每一步都要写磁盘。Hadoop MapReduce 读磁盘 → Map → 写磁盘 → Shuffle → 读磁盘 → Reduce → 写磁盘 每一步都要落盘IO 开销巨大 Spark 读磁盘 → 内存中计算 → 内存中计算 → ... → 必要时写磁盘 能内存算的就内存算飞起官方说 Spark 比 MapReduce 快 100 倍内存计算时虽然实际没这么夸张但快 10~100 倍是常态。3.2 Spark 的核心概念RDDRDDResilient Distributed Dataset弹性分布式数据集是 Spark 的基石# 伪代码Spark RDD 的核心特性classRDD: 1. 分布式数据分散在集群各节点 2. 弹性 lineage血统机制保证容错 3. 不可变转换操作产生新的 RDD而不是修改旧的 defmap(self,func):# 转换Transformation懒执行只记录操作不立即计算returnnew_RDD(lineageself.lineage[func])deffilter(self,func):# 转换Transformation同上returnnew_RDD(lineageself.lineage[func])defcollect(self):# 行动Action触发真正的计算execute_all_transformations()returnresults_to_driver()关键设计特性说明懒加载Lazy Evaluation转换操作不立即执行等到行动操作时才统一计算Lineage 血统记录 RDD 的家谱某个分区丢了可以根据血统重新算内存优先中间结果尽量放内存减少磁盘 IO3.3 Spark 编程比 MapReduce 爽太多了咱们用 Spark 重写前面的单词计数# Spark 版 Word CountPython APIfrompysparkimportSparkContext scSparkContext(local,WordCount)# 1. 读取文件 → RDDlinessc.textFile(hdfs://data/big.txt)# 2. 一系列转换都是懒执行word_counts(lines.flatMap(lambdaline:line.split())# 拆成单词.map(lambdaword:(word,1))# → (word, 1).reduceByKey(lambdaa,b:ab)# 按 key 汇总)# 3. 触发计算Actionresultword_counts.collect()# 4. 输出forword,countinresult:print(f{word}:{count})对比一下 MapReduce 要写几十行 Java 代码Spark 几行 Python 就搞定了而且逻辑清晰得像在写单机程序。3.4 Spark 的生态系统Spark 不只是一个计算引擎它是一整套大数据工具箱组件用途类比Spark SQL用 SQL 分析结构化数据分布式版的 MySQLSpark Streaming实时流处理数据一产生就处理MLlib机器学习库分布式 sklearnGraphX图计算社交网络分析利器# 伪代码Spark SQL 示例frompyspark.sqlimportSparkSession sparkSparkSession.builder.appName(Analysis).getOrCreate()# 直接读 JSON/CSV/Parquet像操作数据库一样dfspark.read.json(hdfs://data/orders.json)# 用 SQLdf.createOrReplaceTempView(orders)resultspark.sql( SELECT user_id, COUNT(*) as cnt, SUM(amount) as total FROM orders WHERE created_at 2022-01-01 GROUP BY user_id )result.show()四、Hadoop vs Spark不是取代而是互补很多人问“有了 SparkHadoop 是不是要被淘汰了”答案是不会它们是搭档。对比维度Hadoop (MapReduce)Spark存储HDFS也可以独立用依赖外部存储HDFS/S3等计算速度慢大量磁盘 IO快内存计算适用场景离线批处理、日志分析迭代计算、实时处理、机器学习容错任务级重算Lineage 血统重算编程难度较繁琐Java 为主简单Python/Scala/Java资源消耗较低内存要求高实际生产中它们通常这样配合┌─────────────────────────────────────────┐ │ 数据存储层 │ │ HDFS / S3 / 对象存储数据仓库 │ └─────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────┐ │ 计算引擎层 │ │ Spark主力处理大部分计算任务 │ │ MapReduce辅助超大规模离线任务 │ └─────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────┐ │ 资源调度层 │ │ YARN / Kubernetes管理集群资源 │ └─────────────────────────────────────────┘HDFS继续当数据仓库存海量数据Spark当计算主力负责大部分分析任务YARN当资源管家协调谁用多少 CPU/内存五、实战从 0 跑一个 Spark 任务光说不练假把式咱们走一遍完整的流程。5.1 环境准备伪代码思路# 1. 启动 HDFSstart-dfs.sh# 2. 启动 YARN资源调度start-yarn.sh# 3. 启动 Spark以 YARN 模式spark-submit\--masteryarn\--deploy-mode cluster\my_app.py5.2 写一个真实的分析任务假设我们要分析用户行为统计每个用户的平均订单金额。# user_behavior_analysis.pyfrompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportavg,count,col# 1. 创建 SparkSession入口sparkSparkSession.builder \.appName(UserBehaviorAnalysis)\.master(yarn)\.getOrCreate()# 2. 读取订单数据存在 HDFS 上ordersspark.read.parquet(hdfs:///data/orders/)# 3. 数据处理user_stats(orders.filter(col(status)completed)# 只算已完成订单.groupBy(user_id)# 按用户分组.agg(count(*).alias(order_count),# 订单数avg(amount).alias(avg_amount)# 平均金额).filter(col(order_count)3)# 只保留有 3 单以上的用户)# 4. 结果写回 HDFS或数据库user_stats.write.mode(overwrite).parquet(hdfs:///output/user_stats/)# 5. 关闭spark.stop()5.3 提交任务# 打包提交到集群spark-submit\--masteryarn\--deploy-mode cluster\--driver-memory 4g\--executor-memory 8g\--executor-cores4\--num-executors10\user_behavior_analysis.py参数说明--driver-memory主控节点的内存--executor-memory每个工作节点的内存--executor-cores每个工作节点用几个 CPU 核--num-executors启动多少个工作者Spark 会根据这些参数把任务分发到集群上并行执行你可以在 Web UI 上看到进度。六、踩坑指南我走过的弯路坑 1内存溢出OOM# 错误示范把海量数据 collect() 到单台机器big_dataspark.read.parquet(10TB_data/)resultbig_data.collect()# boomDriver 内存爆炸# 正确做法结果也写分布式存储big_data.write.parquet(output/)坑 2Shuffle 太多# 错误示范多次 groupBy引发大量 Shuffler1df.groupBy(a).agg(sum(x))r2r1.groupBy(b).agg(avg(y))# 又一次 Shuffle# 优化尽量在一次聚合里完成减少 Shuffle坑 3小文件太多# 如果输入是成千上万个小文件Spark 会起很多任务开销巨大# 解决先合并或用 coalesce/repartitiondf.coalesce(100).write.parquet(output/)七、总结咱们来回顾一下Hadoop解决了大数据怎么存、怎么算的问题是大数据时代的奠基者Spark在 Hadoop 的基础上用内存计算大幅提升了速度让大数据处理更高效实际生产中HDFS Spark YARN是一套经典组合选型时记住离线批处理两者都能做迭代计算和实时处理优先 Spark最后说个感悟大数据技术栈看着庞杂但核心思路始终没变——当一台机器搞不定时就让一群机器一起上。Hadoop 和 Spark 只是这个思路的不同实现理解了这一点再去看 Flink、Kafka 这些也会轻松很多。你在实际项目中用过 Hadoop 或 Spark 吗遇到过什么坑或者对大数据生态有什么疑问欢迎在评论区交流