当前位置：首页 > news >正文

PySpark探索性数据分析：大规模数据勘探实战指南

news 2026/6/15 4:56:41

1. 这不是在Python里跑个df.head()——用PySpark做探索性数据分析的真实战场

“Exploratory Data Analysis (EDA) using PySpark”这个标题，乍看像教科书里的一个章节名，但如果你真在一家日处理TB级日志、千万级用户行为事件、跨多业务线宽表的公司做过数据工作，就会明白：这根本不是“用PySpark重写一遍pandas EDA”，而是一场从思维范式到工程实践的全面切换。我带过三支数据团队，接手过电商、金融风控、IoT设备平台的原始数据湖项目，每次新同学上来第一句都是：“老师，我用pandas写的EDA脚本跑不动了，报了OOM，能不能直接换成PySpark？”。我的回答永远是：“先别换，你得先忘掉pandas那套‘加载-查看-画图’的直觉。”因为PySpark EDA的本质，不是工具迁移，而是数据规模倒逼的认知重构——当你的样本集大到无法单机加载，当你的缺失值统计要扫完200个分区、每个分区3GB，当你的相关系数矩阵计算需要调度12个executor并规避Shuffle爆炸，你做的就不再是“探索”，而是“勘探”：用分布式引擎当钻机，用lazy evaluation当地质雷达，用checkpoint和broadcast变量当安全绳。

核心关键词——Exploratory Data Analysis、PySpark、distributed computing、data profiling、large-scale statistics——已经点明这不是一次轻量级的数据快览，而是在数据湖底进行系统性测绘。它适合三类人：一是刚从传统BI或SQL分析转岗、手握pandas但被生产环境数据量卡住的分析师；二是负责搭建数据质量门禁、需要在ETL前完成字段级健康扫描的工程师；三是正在设计特征工程Pipeline、必须提前摸清分布偏移、异常模式与字段间依赖关系的算法同学。它解决的不是“数据长什么样”的表层问题，而是“这份数据能否支撑下游建模/报表/告警”的底层可信问题。我见过太多项目，因为跳过这一步，在模型上线后才发现主键重复率高达17%，或时间戳字段有38%记录落在未来时区——这些坑，全靠一次扎实的PySpark EDA提前爆破。

这背后的技术逻辑非常硬核：pandas的describe()是内存中对已加载数据的即时计算，而PySpark的summary()是生成一个DAG，触发一次全表Scan，再通过approxQuantile用TDigest算法估算分位数，误差控制在±0.01内；pandas的isnull().sum()是逐行标记再聚合，PySpark则必须用agg(*[count(when(isnull(c), c)).alias(c) for c in df.columns])构造列级聚合表达式，避免广播小表引发Driver OOM；更关键的是，pandas画热力图用seaborn.heatmap(df.corr())一行搞定，PySpark却要先用VectorAssembler把数值列转为向量，再调Correlation.corr()获得稀疏矩阵，最后collect()回Driver——而这一步，就是无数线上任务失败的起点。所以，这篇内容不教你“怎么写”，而教你“为什么必须这么写”，以及“当它崩了，你该盯哪一行日志”。

2. 为什么不能照搬pandas思路？PySpark EDA的四大认知断层与架构设计逻辑

2.1 断层一：从“加载即得”到“计划先行”——Lazy Evaluation如何重塑EDA节奏

pandas用户最深的肌肉记忆，是pd.read_csv("data.csv").head(5)——文件一读，数据立现。PySpark里，spark.read.parquet("s3://bucket/large_table").show(5)看似一样，但背后执行路径天差地别。前者是同步IO+内存解析，后者是异步提交一个Action：Driver先解析SQL Plan，生成Logical Plan，再经Catalyst Optimizer重写（比如把filter().select().limit()合并为单次Scan），最后生成Physical Plan分发给Executor。这意味着，你在PySpark里写的每一行df.filter(...).groupBy(...).count()，都只是在构建DAG节点，真正耗时的计算，只在.show()、.count()、.collect()这些Action触发时才发生。

这个断层直接决定EDA效率。我曾优化过一个金融交易表的初筛流程：原脚本用df.select("amount", "status").filter("status = 'SUCCESS'").describe().show()，耗时47分钟。问题在哪？.describe()默认计算count, mean, stddev, min, max, 25%, 50%, 75%八项统计，但PySpark会为每项单独走一遍全表Scan——相当于扫了8次磁盘。解决方案是改用.summary()，它底层复用同一轮Scan，用approxQuantile批量估算分位数，耗时压到6分钟。更狠的是，我们发现业务只关心amount的分布，于是用df.agg( count("*").alias("total"), count(when(col("status") == "SUCCESS", 1)).alias("success_cnt"), mean("amount").alias("avg_amount"), stddev("amount").alias("std_amount"), approxQuantile("amount", [0.01, 0.25, 0.5, 0.75, 0.99], 0.01) ).show()——把8次Scan压缩成1次，再加approxQuantile的误差容忍，最终耗时112秒。这就是“计划先行”思维的价值：你必须预判哪些统计可合并、哪些必须拆开、哪些能用近似算法替代精确计算。

提示：.summary()的approxQuantile第三个参数是relativeError，设0.01表示误差≤1%，设0.001可压到0.1%，但计算成本翻倍。实测在10亿行数据上，0.01误差下分位数计算比精确percentile_approx快3.2倍，且结果对业务决策无影响。

2.2 断层二：从“列即对象”到“列即表达式”——Column API的不可变性与链式陷阱

pandas里df["age"] = df["age"].fillna(0)是原地修改，PySpark里df.withColumn("age", when(isnull(col("age")), 0).otherwise(col("age")))却是创建新DataFrame。这种不可变性（Immutability）不是设计缺陷，而是为容错服务：每个Transformation都生成新DAG节点，失败时可从最近Checkpoint重放，而非维护复杂状态。但新手常踩的坑，是滥用链式调用导致DAG爆炸。比如分析用户表时，想同时看age缺失率、city唯一值数、reg_time时间范围，有人会写：

df.filter(isnull(col("age"))).count() df.select("city").distinct().count() df.agg(min("reg_time"), max("reg_time")).show()

这看似清晰，实则触发三次全表Scan。正确姿势是单次聚合：

from pyspark.sql.functions import * stats = df.agg( (count(when(isnull(col("age")), 1)) / count("*")).alias("age_null_ratio"), countDistinct("city").alias("city_distinct_cnt"), min("reg_time").alias("min_reg_time"), max("reg_time").alias("max_reg_time") ).collect()[0]

这里的关键洞察是：PySpark的agg()函数接受任意数量的聚合表达式，它们共享同一轮Scan。而countDistinct在Spark 3.0+已优化为HyperLogLog++算法，内存占用恒定O(1)，不像旧版会把所有值Collect到Driver。我在线上环境对比过：对1.2亿用户的city去重，旧版countDistinct因Driver内存溢出失败，新版稳定返回误差<0.8%的结果，耗时仅23秒。

2.3 断层三：从“内存绘图”到“采样导出”——可视化必须绕开Driver瓶颈

pandas EDA的灵魂是matplotlib和seaborn，但PySpark里df.toPandas()是危险操作。当你的表有500列、2亿行，toPandas()会尝试把全部数据Collect到Driver内存，99%的情况是直接OOM。正确的可视化路径是三层过滤：采样→聚合→导出。例如画用户年龄分布直方图：

Step 1：分层采样。不用df.sample(0.01)这种随机采样（可能漏掉稀有年龄段），改用df.stat.sampleBy("age_group", fractions={0:0.05, 1:0.1, 2:0.02}, seed=42)，按业务定义的年龄段分层抽样，保证各区间代表性；
Step 2：本地聚合。在Executor端完成groupBy("age_group").count()，只把几十行聚合结果传回Driver；
Step 3：安全导出。用result_df.coalesce(1).write.mode("overwrite").csv("hdfs://eda_output/age_hist")，让Executor直接写HDFS，Driver零负担。

我曾用这套方法处理一个IoT设备上报表（日增8TB），需监控设备在线时长分布。若用toPandas()，Driver需128GB内存；用分层采样+Executor聚合，仅用16GB Driver内存，3分钟内生成CSV供BI工具读取。记住：PySpark EDA的图表，从来不是在Driver上画的，而是在Executor上算好、存好、再由外部工具渲染的。

2.4 断层四：从“单表快览”到“跨源探查”——Data Profiling的工程化落地

真实业务数据从不孤立存在。一个用户表可能关联着订单表、设备表、营销活动表。pandas时代，我们用pd.merge()拼接后分析；PySpark里，join()是Shuffle重灾区。一次user_df.join(order_df, "user_id")，若user_id分布倾斜（比如某VIP用户有500万订单），会导致单个Task处理GB级数据，拖垮整个Stage。因此，PySpark EDA的顶层设计必须是解耦式探查：先独立分析各表的Profile（字段空值率、唯一值数、值域分布），再用broadcast join处理小维表，最后用skew join技巧处理大表倾斜。

我们为某电商客户设计的探查框架，核心是三个层级：

Level 1：原子表Profile。对每张表运行profile_table(df, columns=None, sample_ratio=0.05)，输出JSON报告，含column_name,null_ratio,distinct_count,min_max,top_k_values（用approx_count_distinct+collect_list实现）；
Level 2：关联健康度。不Join，改用left_anti检查外键引用完整性：“订单表中多少user_id在用户表里找不到？”；
Level 3：业务规则验证。用DataFrameDSL写规则，如df.filter(col("order_amt") < 0).count()查负金额，df.filter(to_date(col("create_time")) > current_date()).count()查未来时间——这些规则可配置化，注入到Airflow DAG自动巡检。

这套设计让客户将数据质量报告生成时间从8小时缩短到22分钟，且报告可直接对接DataHub元数据系统。它的本质，是把EDA从“人工探索”升级为“可调度、可版本化、可告警”的工程能力。

3. 核心实操：从零构建一个生产级PySpark EDA Pipeline（含完整代码与参数详解）

3.1 环境准备与性能基线设定：为什么你的SparkSession配置决定成败

很多人的PySpark EDA卡在第一步：环境没调好。我见过太多案例，不是代码有问题，而是spark.sql.adaptive.enabled没开，或spark.sql.adaptive.skewJoin.enabled关着，导致Shuffle阶段Task耗时差异达100倍。以下是我们在线上集群验证过的最小可行配置（基于Spark 3.3+，YARN模式）：

from pyspark.sql import SparkSession from pyspark.sql.functions import * spark = SparkSession.builder \ .appName("Prod-EDA-Pipeline") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.skewJoin.enabled", "true") \ .config("spark.sql.adaptive.localShuffleReader.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .config("spark.sql.adaptive.broadcastJoinThreshold", "50MB") \ .config("spark.sql.autoBroadcastJoinThreshold", "50MB") \ .config("spark.sql.adaptive.localShuffleReader.maxBufferSize", "128MB") \ .config("spark.sql.files.maxPartitionBytes", "256MB") \ .config("spark.sql.adaptive.skewJoin.skewMapThreshold", "10MB") \ .getOrCreate() # 关键参数解释： # - adaptive.enabled: 开启自适应查询执行（AQE），运行时动态优化DAG # - skewJoin.enabled: AQE自动检测并切分倾斜分区（如把大Key拆成100个小Key） # - coalescePartitions.enabled: 合并小分区，避免过多Task（尤其对小文件多的Parquet表） # - broadcastJoinThreshold: 超过50MB的表不广播，防Driver OOM；小维表可设为100MB # - maxPartitionBytes: 单个分区最大256MB，避免单Task处理过大块 # - skewMapThreshold: Map端倾斜阈值，超10MB即触发AQE倾斜处理

为什么这些参数致命？举个实例：分析一个10TB的用户行为日志表（10万分区），未开AQE时，df.groupBy("event_type").count()因event_type分布极不均衡（"click"占92%，"purchase"占0.3%），导致1个Task耗时42分钟，其余999个Task平均2秒，整体被拖慢。开启AQE后，Spark自动识别click为倾斜Key，将其拆分为click_001到click_100，分散到100个Task，总耗时从42分钟降至3分17秒。这就是配置即生产力。

注意：spark.sql.adaptive.enabled必须设为true，否则后续所有AQE参数无效。我们曾因配置文件里写成"True"（首字母大写），导致AQE静默失效，排查了两天日志。

3.2 原子表深度探查：字段级Profile的七维诊断法

我们定义的profile_column函数，不是简单统计，而是七维穿透式诊断。以user_id字段为例：

def profile_column(df, col_name, sample_ratio=0.1): """ 对单列进行七维诊断： 1. null_ratio: 空值率（用approx_count_distinct规避count(*)全扫） 2. distinct_ratio: 唯一值占比（评估主键/索引质量） 3. min_max: 数值型字段的极值（字符串型取长度极值） 4. top_k_values: 频次最高的5个值及占比（用collect_list+map_reduce模拟） 5. data_type: 推断实际类型（string/int/float/timestamp） 6. pattern_match: 正则匹配常见模式（如手机号、邮箱、UUID） 7. skewness: 分布偏度（用skewness()函数，仅数值型） """ from pyspark.sql.types import * # Step 1: 基础统计（单次Scan） base_stats = df.select( count("*").alias("total_cnt"), count(when(col(col_name).isNull() | isnan(col(col_name)), 1)).alias("null_cnt"), countDistinct(col(col_name)).alias("distinct_cnt"), approxCountDistinct(col(col_name), rsd=0.01).alias("approx_distinct_cnt") # RSD=0.01即误差1% ).collect()[0] null_ratio = base_stats["null_cnt"] / base_stats["total_cnt"] distinct_ratio = base_stats["approx_distinct_cnt"] / base_stats["total_cnt"] # Step 2: 类型推断与极值（需根据schema判断） col_dtype = [f.dataType for f in df.schema.fields if f.name == col_name][0] if isinstance(col_dtype, (IntegerType, LongType, DoubleType, FloatType)): ext_stats = df.agg( min(col(col_name)).alias("min_val"), max(col(col_name)).alias("max_val"), skewness(col(col_name)).alias("skewness"), stddev(col(col_name)).alias("stddev") ).collect()[0] min_max = f"{ext_stats['min_val']}-{ext_stats['max_val']}" skewness_val = ext_stats["skewness"] else: # 字符串型：取长度极值 len_stats = df.agg( min(length(col(col_name))).alias("min_len"), max(length(col(col_name))).alias("max_len") ).collect()[0] min_max = f"{len_stats['min_len']}-{len_stats['max_len']}" skewness_val = None # Step 3: Top 5 values（用采样+本地聚合规避collect全量） sampled_df = df.sample(fraction=sample_ratio, seed=42) top5 = sampled_df.groupBy(col(col_name)).count() \ .orderBy(desc("count")) \ .limit(5) \ .rdd.map(lambda r: (r[0], r[1] / (base_stats["total_cnt"] * sample_ratio))) \ .collect() # 采样后归一化，误差可控 # Step 4: 模式匹配（预定义正则） patterns = { "phone": r"^1[3-9]\d{9}$", "email": r"^[^\s@]+@[^\s@]+\.[^\s@]+$", "uuid": r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" } pattern_match = {} for name, pat in patterns.items(): cnt = df.filter(col(col_name).rlike(pat)).count() pattern_match[name] = f"{cnt}/{base_stats['total_cnt']} ({cnt/base_stats['total_cnt']:.2%})" return { "column": col_name, "null_ratio": round(null_ratio, 4), "distinct_ratio": round(distinct_ratio, 4), "min_max": min_max, "top_5_values": top5, "data_type": str(col_dtype), "pattern_match": pattern_match, "skewness": skewness_val } # 实际调用 profile_result = profile_column(user_df, "user_id", sample_ratio=0.05) print(profile_result)

这段代码的核心价值在于误差可控的采样策略：top_5_values用sample(0.05)后聚合，再除以(total_cnt * 0.05)归一化，实测在10亿行数据上，top1值占比误差<0.3%；而approxCountDistinct用RSD=0.01，内存占用仅为精确countDistinct的1/15。我们曾用此方法扫描一个200列的用户宽表，全程耗时8.3分钟，生成的JSON报告包含所有字段的健康画像，成为数据治理平台的输入源。

3.3 关联表健康度探查：不Join的外键完整性验证

跨表分析是EDA的高危区。我们坚持“能不Join，就不Join”原则，用集合运算替代连接。以验证订单表order_df的user_id是否都在用户表user_df中为例：

def check_foreign_key_integrity(parent_df, child_df, parent_key, child_key, sample_ratio=0.01, threshold=0.995): """ 验证外键完整性：child表中child_key有多少比例在parent表的parent_key中存在 不使用join，改用left_anti + sample提升速度 """ # Step 1: 对child表采样，减少计算量 sampled_child = child_df.sample(fraction=sample_ratio, seed=42) # Step 2: 找出sampled_child中不在parent中的记录（left_anti） missing_ids = sampled_child.join( parent_df.select(col(parent_key).alias("p_key")), sampled_child[child_key] == col("p_key"), "left_anti" ) # Step 3: 计算缺失率 total_sampled = sampled_child.count() missing_cnt = missing_ids.count() missing_ratio = missing_cnt / total_sampled if total_sampled > 0 else 0 # Step 4: 推断全量缺失率（置信区间） from math import sqrt se = sqrt(missing_ratio * (1 - missing_ratio) / total_sampled) # 标准误 ci_lower = max(0, missing_ratio - 1.96 * se) # 95%置信下限 result = { "parent_table": parent_df._jdf.toString(), "child_table": child_df._jdf.toString(), "parent_key": parent_key, "child_key": child_key, "sample_ratio": sample_ratio, "missing_ratio": round(missing_ratio, 4), "ci_lower_95%": round(ci_lower, 4), "is_valid": ci_lower <= (1 - threshold) # 缺失率下限低于阈值才认为OK } return result # 调用示例 fk_check = check_foreign_key_integrity( user_df, order_df, "id", "user_id", sample_ratio=0.02, threshold=0.998 ) print(fk_check) # 输出：{'missing_ratio': 0.0012, 'ci_lower_95%': 0.0008, 'is_valid': True}

这个方案的精妙之处在于用统计推断替代全量计算。对10亿行订单表，全量left_anti需扫描全部数据，耗时约15分钟；而采样2%，仅需扫描2000万行，耗时48秒，且95%置信下限0.0008意味着“全量缺失率有95%概率≤0.08%”，完全满足业务对数据一致性的要求。我们把它封装成Airflow Operator，每天凌晨自动校验核心表关联，异常时钉钉告警。

3.4 业务规则探查：可配置化的数据质量守门员

真正的EDA必须嵌入业务语义。我们设计了一个RuleEngine类，支持YAML规则配置：

# rules.yaml rules: - name: "negative_order_amount" description: "订单金额不能为负" expression: "order_amt < 0" severity: "critical" - name: "future_create_time" description: "创建时间不能晚于当前时间" expression: "create_time > current_timestamp()" severity: "warning" - name: "user_age_range" description: "用户年龄应在0-120之间" expression: "age < 0 OR age > 120" severity: "error" # RuleEngine实现 class RuleEngine: def __init__(self, spark): self.spark = spark def validate_rules(self, df, rules_yaml_path): import yaml with open(rules_yaml_path) as f: config = yaml.safe_load(f) results = [] for rule in config["rules"]: try: # 动态解析expression为Column对象 expr = eval(rule["expression"]) violation_cnt = df.filter(expr).count() # 计算违规率 total_cnt = df.count() violation_ratio = violation_cnt / total_cnt if total_cnt > 0 else 0 results.append({ "rule_name": rule["name"], "description": rule["description"], "violation_count": violation_cnt, "violation_ratio": round(violation_ratio, 6), "severity": rule["severity"], "status": "FAILED" if violation_cnt > 0 else "PASSED" }) except Exception as e: results.append({ "rule_name": rule["name"], "error": str(e), "status": "ERROR" }) return results # 使用 engine = RuleEngine(spark) report = engine.validate_rules(order_df, "rules.yaml") for r in report: print(f"{r['rule_name']}: {r['status']} ({r.get('violation_ratio', 'N/A')})")

这个设计让业务方能自主维护规则，无需改动代码。我们曾用它发现某支付渠道因时区配置错误，导致23%的订单create_time被写成未来时间，及时拦截了下游报表污染。关键点在于eval(rule["expression"])——它把字符串表达式安全地转为PySpark Column，比硬编码df.filter(col("order_amt") < 0)灵活百倍，且支持current_timestamp()等动态函数。

4. 血泪教训：PySpark EDA中那些没人告诉你的12个致命坑与避坑指南

4.1 坑1：`df.count()`是EDA最大杀手，90%的超时源于此

新手第一反应是df.count()看行数，但在TB级数据上，这是最昂贵的Action。count()必须走全表Scan，且无法利用任何谓词下推（Predicate Pushdown）。我们曾有个日志表，df.count()耗时37分钟，而业务其实只需要知道“是否大于1亿行”。解决方案是用df.selectExpr("count(1) as cnt").limit(1).collect()——Spark会优化为只计算count，不物化中间数据；更激进的是，用df.inputFiles()获取文件列表，估算行数：“S3上1200个Parquet文件，平均每个200MB，按每GB 1000万行估算，总量≈24亿行”，误差±15%，但耗时<3秒。

实操心得：在EDA脚本开头，永远先用df.printSchema()和df.explain("formatted")看执行计划，如果Plan里有WholeStageCodegen但没Filter下推，说明你的filter条件没生效，count()就是在裸扫。

4.2 坑2：`toPandas()`不是万能钥匙，它是Driver内存的定时炸弹

df.toPandas()在小数据上很香，但一旦数据量超过Driver内存的1/3，必然OOM。我们线上Driver配32GB，安全阈值是10GB数据。破解之道是分块导出：

def safe_to_pandas(df, max_rows=5000000): """安全导出：按分区切分，逐块collect""" num_partitions = df.rdd.getNumPartitions() rows_per_partition = max_rows // num_partitions + 1 # 重分区确保每块不超过rows_per_partition repartitioned = df.repartitionByRange(num_partitions, "id") # 按主键范围重分区 blocks = [] for i in range(num_partitions): block = repartitioned.rdd.mapPartitionsWithIndex( lambda idx, it: it if idx == i else [] ).toDF(df.schema) try: pandas_block = block.limit(rows_per_partition).toPandas() blocks.append(pandas_block) print(f"Block {i} collected, shape: {pandas_block.shape}") except Exception as e: print(f"Block {i} failed: {e}") break return pd.concat(blocks, ignore_index=True) if blocks else pd.DataFrame() # 使用 sample_df = safe_to_pandas(user_df, max_rows=1000000)

这个函数用repartitionByRange按主键排序分块，确保每块数据均匀，再逐块toPandas()。我们用它成功导出过800万行用户样本，耗时2分18秒，Driver内存峰值仅11GB。

4.3 坑3：`approxQuantile`的精度陷阱——为什么0.01误差有时不够

approxQuantile的relativeError参数，很多人以为设0.01就是误差1%，但实际是相对分位数值的误差。比如求99%分位数，真实值是10000，relativeError=0.01允许返回9900~10100；但如果真实值是100，同样误差下只允许99~101。在监控指标时，这可能导致误判。我们的解决方案是双精度校验：对关键分位数（如95%、99%），先用approxQuantile(..., 0.01)快速获取，再对结果附近的值做精确扫描：

# 先用近似获取粗略值 q95_approx = df.approxQuantile("response_time", [0.95], 0.01)[0] # 再在[q95_approx*0.9, q95_approx*1.1]范围内精确计算 precise_range = df.filter( (col("response_time") >= q95_approx * 0.9) & (col("response_time") <= q95_approx * 1.1) ) q95_precise = precise_range.approxQuantile("response_time", [0.95], 0.001)[0]

这样既保证速度，又守住精度。实测在10亿行响应时间数据上，双精度方案比纯approxQuantile(..., 0.001)快4.7倍。

4.4 坑4：`collect_list`的内存黑洞——Top N值统计的优雅解法

df.groupBy("category").agg(collect_list("product_id")).show()看似简单，但collect_list会把所有值Collect到单个Task，极易OOM。正确姿势是用array_sort+slice组合：

from pyspark.sql.functions import * # 获取每个category下销量Top 10的product_id top10_per_cat = df.groupBy("category").agg( array_sort( collect_list(struct("product_id", "sales_cnt")), lambda x, y: y["sales_cnt"] - x["sales_cnt"] # 降序排列 ).alias("sorted_products") ).select( "category", slice("sorted_products", 1, 10).alias("top10_products") # 取前10 )

array_sort在Executor端完成排序，slice只取前10，全程不Collect全量。我们用此法处理过一个15亿行的商品销售表，Top 10统计耗时3分42秒，内存占用稳定在8GB以内。

4.5 坑5：时间字段的时区幻觉——`current_date()`vs`current_timestamp()`

df.filter(col("create_time") > current_date())看似合理，但current_date()返回的是UTC日期，而你的数据可能是东八区时间。结果就是，所有当天16点后的记录都被误判为“未来时间”。必须统一时区：

# 正确做法：显式指定时区 from pyspark.sql.functions import * # 将create_time转为东八区，再与当前东八区时间比较 df.filter( to_utc_timestamp(col("create_time"), "Asia/Shanghai") > to_utc_timestamp(current_timestamp(), "Asia/Shanghai") ) # 或更简单：用localtimestamp()（返回Driver本地时区时间） df.filter(col("create_time") > localtimestamp())

我们曾因这个Bug，让一个实时风控模型误杀了37%的正常交易，根源就是current_date()的时区不匹配。

4.6 坑6：`broadcast`变量的隐形泄漏——为什么你的Driver内存越用越多

broadcast变量本意是把小表分发到Executor，但若在循环中反复spark.sparkContext.broadcast()，旧的broadcast不会自动GC，导致Driver内存泄漏。我们的规范是：broadcast变量必须命名，并在使用后显式销毁：

# 正确：命名+销毁 city_map_bc = spark.sparkContext.broadcast(city_dict) try: result_df = df.withColumn("city_name", lookup_city_udf(col("city_id"))) finally: city_map_bc.unpersist() # 显式释放 # 错误：匿名broadcast，无法追踪 spark.sparkContext.broadcast(large_dict) # 内存永不释放

线上集群监控显示，规范使用broadcast后，Driver内存波动从±15GB降至±2GB。

4.7 坑7：`cache()`的甜蜜陷阱——缓存什么、何时缓存、缓存多久

df.cache()不是万能加速器。对只用一次的DataFrame缓存，反而增加序列化开销。我们的缓存策略是“三不缓存”：不缓存只用一次的、不缓存小表（<1GB）、不缓存即将被filter大幅裁剪的。最佳实践是df.persist(StorageLevel.MEMORY_AND_DISK_SER)，序列化后存内存+磁盘，OOM时自动落盘。更关键的是，缓存后必须count()触发，否则只是声明，不执行：

# 错误：声明了但没触发 df_cached = df.filter("status = 'ACTIVE'").cache() # ... 后面直接用df_cached，但没count，缓存未生效 # 正确：声明+触发 df_cached = df.filter("status = 'ACTIVE'").cache() df_cached.count() # 强制触发缓存

我们曾优化一个报表脚本，去掉无意义的cache()，耗时从21分钟降至14分钟。