1. 从WordCount入门Spark核心算子第一次接触Spark时WordCount就像编程界的Hello World简单却能完整展示Spark的思维方式。记得我刚开始用Spark处理日志分析时面对每天TB级的Nginx日志传统单机脚本跑几个小时都完不成而改写为WordCount模式后集群十分钟就能出结果。先看这个经典案例的需求统计文本中每个单词的出现频率并按次数降序排列。用Python原生写法可能需要几十行循环和字典操作而Spark版本只需要4个核心算子的链式调用sc SparkContext(local, pySpark) rdd sc.textFile(/root/wordcount.txt) result rdd.flatMap(lambda x: str(x).split( )) \ .map(lambda x: (x, 1)) \ .reduceByKey(lambda x, y: x y) \ .sortBy(lambda x: x[1], False)这串代码就像工厂的流水线每个算子都是个智能工人textFile搬运工把HDFS/本地文件拆分成多个数据块装载到RDD这个分布式集装箱里flatMap拆箱专家把每行文本拆成单词注意与map的区别flatMap会压平嵌套结构map标签员给每个单词贴上(单词,1)的标签reduceByKey分类计数员按单词分组累加次数sortBy排序师按词频降序排列我曾用这个模式处理过微博热搜词统计当数据量从MB增长到TB级时只需增加集群节点代码完全不用修改——这就是Spark的魔力。但要注意两个性能陷阱数据倾斜像的、是这类高频词会导致某些节点负载过高可以先用sample算子检查数据分布shuffle开销reduceByKey会触发网络传输对小数据集可用collect()本地排序替代2. 解密Spark算子的执行机制很多初学者会困惑为什么Spark代码看着像普通Python程序却能分布式执行这要归功于RDD弹性分布式数据集的巧妙设计。去年优化一个推荐系统时我通过理解算子执行原理把作业速度提升了3倍。Spark算子分为两大类转换算子Transformation如map、filter它们像菜谱步骤只记录操作不立即执行行动算子Action如collect、count像点火开关触发实际计算这种惰性求值机制让Spark可以优化整个执行计划。比如下面这段代码logs sc.textFile(hdfs://logs/*) errors logs.filter(lambda x: ERROR in x) warnings logs.filter(lambda x: WARN in x) error_count errors.count() warning_count warnings.count()虽然有两个filter操作但Spark会智能地只扫描一次HDFS文件。我曾遇到个典型问题某个ETL作业读取相同数据源十几次通过persist()缓存RDD后运行时间从2小时降到15分钟raw_data sc.textFile(hdfs://data/*).persist() # 缓存到内存 process1(raw_data) process2(raw_data) # 直接使用缓存关键原理窄依赖像map这类算子分区之间数据独立无需网络传输宽依赖如groupByKey需要跨节点混洗shuffle数据血统LineageRDD记录构建过程丢失数据时可快速重建看个实际性能对比表算子组合是否触发shuffle适合场景map - filter否数据清洗groupByKey - mapValues是分组统计repartition - map是数据重平衡3. 好友推荐实战复杂算子组合从WordCount升级到社交网络分析就像从单车换成赛车。去年为某社交平台开发推荐系统时我们需要计算用户之间的潜在好友关系这正是共同好友算法的用武之地。假设数据格式为用户 好友1 好友2...要推荐非直接好友但共同好友多的人。算法分三步生成所有直接好友对找出共同好友按共同好友数排序def relations(items): result [] for i in range(1, len(items)): result.append((f{items[0]}_{items[i]}, 0)) # 直接好友标记为0 for j in range(i1, len(items)): result.append((f{items[i]}_{items[j]}, 1)) # 潜在好友标记为1 return result rdd sc.textFile(/root/friend.txt) \ .map(lambda x: x.split()) \ .flatMap(relations) \ .reduceByKey(lambda x, y: 0 if x0 or y0 else xy) \ .filter(lambda x: x[1]0)这个案例用到了几个高阶技巧嵌套数据结构处理relations函数返回列表的列表flatMap将其展平自定义标记策略用0/1区分直接/间接关系二次过滤reduceByKey后剔除直接好友关系实际项目中我们还加入了这些优化权重计算共同好友的亲密度不同如互动频率黑名单过滤已拉黑用户不推荐性能监控用Spark UI观察shuffle数据量# 带权重的进阶版 def weighted_relations(items): for i in range(1, len(items)): yield (f{items[0]}_{items[i]}, (0, weight[items[0]][items[i]])) for j in range(i1, len(items)): yield (f{items[i]}_{items[j]}, (1, min(weight[items[i]][items[0]], weight[items[j]][items[0]]))) # 在reduce阶段同时处理标记和权重 .reduceByKey(lambda x, y: (0, 0) if x[0]0 or y[0]0 else (1, x[1]y[1]))4. 生产环境调优经验在电商公司处理用户行为日志时我踩过不少性能坑总结出这些实战经验资源分配黄金法则每个CPU核心分配2-4个任务执行器内存 (总内存 - 1GB) / 节点数避免单个分区超过2GB会触发Spark的溢出机制conf SparkConf() \ .set(spark.executor.memory, 4g) \ .set(spark.executor.cores, 2) \ .set(spark.default.parallelism, 200)常见问题解决方案数据倾斜# 方法1加盐处理 skewed_rdd rdd.map(lambda x: (x str(random.randint(0,9)), 1)) # 方法2两阶段聚合 stage1 rdd.map(lambda x: (x, 1)).reduceByKey(lambda x,y: xy) stage2 stage1.map(lambda x: (x[1], x[0])).sortByKey(False)小文件问题# 读取时合并 df spark.read.option(mergeSchema, true).parquet(/path/*) # 输出时控制 df.coalesce(10).write.parquet(/output/)缓存策略选择MEMORY_ONLY默认策略性能最好但可能溢出MEMORY_AND_DISK内存不足时溢写到磁盘DISK_ONLY超大且不频繁访问的数据监控指标解读Scheduler Delay 500ms 表示资源不足Shuffle Read Size突然增大可能需调整分区数GC Time占比过高需要优化JVM参数最后分享一个真实案例某次大促期间推荐作业突然从30分钟暴增到2小时。通过Spark UI发现99%的任务在1分钟内完成但有3个任务跑了1.5小时——这就是典型的数据倾斜。最终通过采样定位热点key - 分离处理 - 结果合并的三步方案解决了问题。