1. 项目概述为什么在大数据场景下EDA不能再只靠Pandas了“Exploratory Data Analysis (EDA) using Pyspark”——这个标题乍看平平无奇但背后藏着一个几乎所有数据工程师、分析型数据科学家在真实业务中都会撞上的墙当你的数据量从百万行跳到十亿行当本地笔记本跑df.describe()要卡住三分钟当df.hist()直接报MemoryError你才真正意识到传统EDA那套“加载→清洗→可视化→推断”的流水线在大数据面前不是慢而是根本走不通。我做过不下20个跨行业数据平台迁移项目最常听到的抱怨就是“模型代码写完了但连数据长什么样都没搞清楚。”不是不想做EDA是Pandas一读就崩Dask调参像玄学而Spark——尤其是PySpark——恰恰是唯一能把“探索”这件事重新拉回工程化轨道的工具。核心关键词“PySpark”“EDA”“大数据探索”不是技术堆砌而是现实妥协后的最优解它不追求单机可视化那种像素级细腻但能保证你在TB级原始日志里5分钟内确认字段分布是否异常、空值率是否突变、时间戳是否全为1970、用户ID是否突然出现超长字符串——这些才是上线前真正要命的问题。它适合三类人一是刚从BI或Python分析岗转来的大数据团队成员需要把原有分析思维迁移到分布式环境二是数据平台建设者必须为下游建模团队提供可信的数据质量快照三是MLOps工程师要在特征工程前完成可复现、可审计的探索流程。这不是教你怎么画漂亮的箱线图而是教你用pyspark.sql.DataFrame原生API在不落地、不采样失真、不触发Shuffle的前提下拿到真正有业务意义的统计信号。下面所有内容都来自我在电商实时风控、金融反欺诈、IoT设备日志三个典型场景中反复打磨出的实操路径。2. 整体设计思路避开三大经典陷阱构建可持续的EDA工作流2.1 为什么不能直接把Pandas EDA脚本改成PySpark——分布式语义的根本差异很多团队第一步就踩坑把原来用pandas.DataFrame写的df.isnull().sum()、df.groupby(category).size()直接替换成spark_df.agg(...)结果发现耗时翻了5倍集群资源打满还经常OOM。问题不在代码转换而在思维没切换。Pandas是“数据在内存计算在CPU”所有操作都是即时执行PySpark是“数据在存储计算在Executor”每一步.agg()、.filter()都只是生成DAG有向无环图真正的执行要等到.show()、.count()或.collect()这种Action触发。更关键的是Pandas的describe()返回的是完整数值矩阵而PySpark的describe()默认只对数值列做count, mean, stddev, min, max且min/max在大数据下可能因分区不均产生偏差——我曾在某物流轨迹数据上发现spark_df.describe().show()返回的max比实际最大值小了37%因为某个分区恰好没分到极端离群点。所以我的设计原则第一条拒绝“翻译式迁移”坚持“语义重构”。比如Pandas里一句df[price].hist(bins50)在PySpark里没有等价API强行用toPandas().hist()等于放弃分布式优势。正确做法是用approxQuantile获取分位数结合bucketBy做近似直方图分桶再用agg统计各桶频次——虽然少了曲线平滑度但保证了TB级数据下5秒内返回结果且误差可控后文会给出误差计算公式。2.2 为什么不用Databricks AutoML或AWS Deequ——轻量级与可控性的权衡市面上已有不少封装好的EDA工具比如Databricks的display(spark_df)自动渲染统计面板或Deequ的VerificationSuite做数据质量校验。它们确实省事但有两个硬伤第一深度定制难。比如你要分析“用户首次点击到下单的时间间隔分布”这需要自定义UDF用户定义函数计算时间差而AutoML的统计面板根本不支持嵌套逻辑第二调试黑盒化。当VerificationSuite.run()报出“completeness: 0.92”时你无法快速定位是哪张表、哪个分区、哪类用户导致缺失——而原生PySpark的.filter(col(event_time).isNull()).select(user_id, page_url).show(10)能立刻看到问题样本。在我的金融客户项目中正是靠手动写aggfilter组合发现了某支付渠道的transaction_id字段在凌晨2-4点批量为空根源是第三方SDK心跳包异常这种根因定位能力是黑盒工具给不了的。因此整体架构采用“三层漏斗”设计第一层Schema与基础元数据扫描毫秒级检查列名、数据类型、nullable标记识别string列中是否混入数字如123和123并存第二层轻量级统计探针秒级用approxQuantile替代min/max用sample(withReplacementFalse, fraction0.01)做无偏采样避免全表扫描第三层按需深度探查分钟级针对可疑列启动带WHERE条件的定向聚合如spark_df.filter(col(amount) 0).agg(count(*), avg(amount)).show()。这个结构确保90%的日常检查在10秒内完成只有真正需要深挖时才消耗集群资源。2.3 为什么强调“可复现性”而非“可视化美观”——生产环境的生存法则在Kaggle比赛里一张Seaborn热力图能帮你拿奖但在银行风控系统里一张热力图连上线评审都过不了。生产环境要求的是同一份数据今天跑和明天跑结果一致同一段代码A同事跑和B同事跑输出相同统计口径变更时能快速定位影响范围。PySpark天然支持这些——.cache()保证多次计算用同一份缓存.checkpoint()切断DAG依赖DataFrame.explain(modeformatted)能打印物理执行计划。而Matplotlib/Plotly生成的图片既无法版本化管理也无法自动化比对。所以我所有EDA脚本的输出都是结构化DataFramestats_summary spark_df.agg(*[...]).toPandas()然后导出CSV存入数据治理平台。某次客户审计时他们直接用这份CSV和上游ETL日志做diff30分钟就确认了数据漂移发生在哪个调度任务这种确定性是任何可视化工具换不来的。3. 核心细节解析12个必掌握的PySpark EDA操作与避坑指南3.1 Schema诊断从printSchema()到analyze_schema()的升级spark_df.printSchema()只能看树状结构但真实数据中常见“类型污染”比如user_age声明为IntegerType但实际值包含N/A、、unknown。这类问题printSchema()完全不报却会导致后续agg(avg(user_age))直接失败。我的解决方案是写一个analyze_schema()函数from pyspark.sql.types import * from pyspark.sql.functions import * def analyze_schema(df): # 获取每列的原始类型和非空计数 type_counts [] for field in df.schema.fields: col_name field.name col_type str(field.dataType) # 统计该列中各数据类型的实例数需先cast尝试 if string in col_type.lower(): # 对string列尝试转int/float/bool统计成功数 int_count df.filter(col(col_name).rlike(r^-?\d$)).count() float_count df.filter(col(col_name).rlike(r^-?\d*\.?\d$)).count() bool_count df.filter(lower(col(col_name)).isin([true,false])).count() null_count df.filter(col(col_name).isNull() | (col(col_name) )).count() total_count df.count() type_counts.append({ column: col_name, declared_type: col_type, int_ratio: int_count / total_count if total_count 0 else 0, float_ratio: float_count / total_count if total_count 0 else 0, bool_ratio: bool_count / total_count if total_count 0 else 0, null_or_empty_ratio: null_count / total_count if total_count 0 else 0 }) return spark.createDataFrame(type_counts)这个函数会输出类似这样的结果columndeclared_typeint_ratiofloat_ratiobool_rationull_or_empty_ratiouser_ageStringType0.820.030.000.15is_premiumStringType0.000.000.940.06提示rlike正则匹配比cast更安全因为cast失败会静默转为null而rlike只匹配模式不会改变数据。实测在10亿行用户表上这个分析耗时12秒比全表cast快8倍。3.2 数值列分布探查approxQuantile的精度控制与误差边界Pandas的df[amount].quantile(0.95)是精确计算但PySpark的spark_df.approxQuantile(amount, [0.95], 0.01)是近似算法第三个参数relativeError相对误差决定精度。很多人设成0.001以为更准结果发现耗时暴增——因为Spark内部用的是GK算法Greenwald-KhannarelativeError0.001意味着要维护约1000个概要点每个点都要跨分区合并网络开销极大。我的经验是对EDA场景relativeError0.02即2%误差完全够用。比如amount真实95分位是¥298.5approxQuantile返回¥292~¥305都属合理范围毕竟EDA目标是发现数量级异常如95分位突然从¥300跳到¥3000不是做财务审计。计算误差边界的公式很简单绝对误差 ≈ relativeError × (max - min)所以如果已知amount范围是¥0~¥10000设relativeError0.02则绝对误差上限是¥200。这意味着只要观察到95分位变化超过¥500就一定是真实漂移不是算法误差。我在电商大促监控中就用这个逻辑当approxQuantile(order_amount, [0.95], 0.02)从¥298跳到¥850立即触发告警人工确认是新上线的“满1000减500”活动导致而非数据问题。3.3 类别列高频值提取freqItems的隐藏参数与采样陷阱spark_df.freqItems([category], support0.01)能找出出现频率≥1%的类别值但默认行为很坑它会对整张表做全量扫描且support参数是全局阈值。如果表有10亿行support0.01意味着要找出现≥1000万次的值而实际业务中“高频”可能是“在最近7天数据中占10%”。我的改进是加两层过滤# 先按时间分区采样假设表有date_partition列 recent_df spark_df.filter(col(date_partition) 2024-01-01) # 再用sample降低基数 sampled_df recent_df.sample(False, 0.05) # 5%无放回采样 # 最后求高频项 freq_items sampled_df.freqItems([category], support0.1) # 10%阈值基于采样后数据这里的关键是sample(False, 0.05)的False表示无放回避免同一高危样本被重复采样放大偏差support0.1是针对采样后数据的阈值实际对应原始数据的0.0050.1×0.05但计算更快。某次在广告点击日志中用此法5秒内找到TOP10曝光频道而全表freqItems跑了17分钟。3.4 时间列健康度检查date_format与unix_timestamp的组合杀招时间列是数据质量重灾区格式混乱2024-01-01vs01/01/2024、时区错乱UTC存成CST、粒度不一2024-01-01 10:00:00vs2024-01-01。Pandas用pd.to_datetime()能自动容错PySpark必须显式指定格式。我的标准检查流程from pyspark.sql.functions import * # 步骤1尝试用最严格格式解析ISO标准 parsed_ts to_timestamp(col(event_time), yyyy-MM-dd HH:mm:ss.SSS) # 步骤2标记解析失败的行 is_valid ~col(event_time).isNull() parsed_ts.isNotNull() # 步骤3统计失败率 invalid_ratio spark_df.agg( (1 - avg(is_valid.cast(double))).alias(invalid_ratio) ).collect()[0][invalid_ratio] # 步骤4对有效时间检查是否在合理范围内如不早于2020年不晚于当前时间1天 valid_times spark_df.filter(is_valid) time_range_check valid_times.agg( min(parsed_ts).alias(min_time), max(parsed_ts).alias(max_time), count(when(parsed_ts 2020-01-01, 1)).alias(too_old_count), count(when(parsed_ts date_add(current_date(), 1), 1)).alias(too_new_count) ).collect()[0]注意to_timestamp第二个参数必须是Spark支持的Java SimpleDateFormat不能用%Y-%m-%d这种Python风格。我曾因写错格式串导致整个时间列解析全为null排查了2小时才发现是YYYY周年度和yyyy年度的区别。3.5 空值与零值联合分析超越isNull()的业务语义识别isNull()只能抓SQL意义上的NULL但业务中还有大量“伪空值”空字符串、N/A、0金额为0可能代表未支付、-1状态码-1代表未知。我的做法是定义“业务空值字典”然后用when/otherwise统一标记from pyspark.sql.functions import * # 定义业务空值映射根据领域调整 business_null_map { user_id: [N/A, NULL, ], amount: [0, -1], status: [UNKNOWN, PENDING] } null_flags [] for col_name, null_values in business_null_map.items(): if null_values: if isinstance(null_values[0], str): # 字符串匹配 condition reduce(lambda a, b: a | (col(col_name) b), null_values, lit(False)) else: # 数值匹配 condition reduce(lambda a, b: a | (col(col_name) b), null_values, lit(False)) null_flags.append( when(condition | col(col_name).isNull(), 1).otherwise(0).alias(f{col_name}_is_business_null) ) # 合并标记 flagged_df spark_df.select(*, *null_flags) # 统计各列业务空值率 null_stats flagged_df.agg(*[ avg(f{col_name}_is_business_null).alias(f{col_name}_business_null_ratio) for col_name in business_null_map.keys() ])这个方法让我在某保险理赔数据中发现claim_amount列有12%的0值但其中8%是“预估金额未定”2%是“拒赔”2%是真实0赔付——如果不区分直接删空值会丢失关键业务状态。3.6 数据倾斜初筛groupBy前的sample与approxCountDistinctspark_df.groupBy(user_id).count()是检测倾斜的经典方法但如果user_id有10亿唯一值这个Job会因Shuffle爆炸而失败。我的轻量级筛查法# 步骤1用approxCountDistinct估算唯一值总数 total_distinct spark_df.agg(approx_count_distinct(user_id)).collect()[0][0] # 步骤2采样1%数据做局部groupBy sampled_group spark_df.sample(False, 0.01).groupBy(user_id).count() # 步骤3统计采样中出现频次100的user_id即采样中已显现出高度集中 skewed_candidates sampled_group.filter(col(count) 100).select(user_id) # 步骤4用broadcast join反查原表确认这些候选者的实际频次 # 避免全表join只查可疑ID candidate_list [row[user_id] for row in skewed_candidates.collect()] if candidate_list: broadcast_df spark.sparkContext.broadcast(candidate_list) def is_skewed_user(user_id): return user_id in broadcast_df.value is_skewed_udf udf(is_skewed_user, BooleanType()) confirmed_skew spark_df.filter(is_skewed_udf(col(user_id))).groupBy(user_id).count()这套组合拳能在30秒内锁定Top10倾斜Key比全量groupBy快200倍。某次在用户行为日志中发现user_idSYSTEM占了总流量的43%根源是埋点SDK的默认兜底值及时推动前端修复。3.7 字符串列内容探查length、regexp_extract与levenshtein的实战组合字符串列最难搞因为长度、编码、特殊字符都可能引发问题。我的标准化检查清单from pyspark.sql.functions import * # 1. 长度分布避免超长文本拖慢处理 len_stats spark_df.agg( min(length(col(product_name))).alias(min_len), max(length(col(product_name))).alias(max_len), avg(length(col(product_name))).alias(avg_len), stddev(length(col(product_name))).alias(std_len) ) # 2. 编码异常检测含不可见字符 # 检查是否含制表符、换行符、零宽空格 has_control_chars ( col(product_name).rlike(r[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]) | col(product_name).contains(\u200B) # 零宽空格 ) control_char_ratio spark_df.agg(avg(has_control_chars.cast(double))).collect()[0][0] # 3. 关键词提取如从描述中抽品牌名 # 用regexp_extract匹配常见品牌正则 brand_pattern r(apple|samsung|huawei|xiaomi) extracted_brands spark_df.withColumn( brand, regexp_extract(col(product_name), brand_pattern, 0) ).filter(col(brand) ! ).groupBy(brand).count().orderBy(desc(count)) # 4. 相似度聚类发现拼写错误 # 计算Levenshtein距离找编辑距离≤2的相似字符串 from pyspark.sql.functions import levenshtein similar_pairs spark_df.alias(a).crossJoin(spark_df.alias(b)) \ .filter(col(a.product_id) ! col(b.product_id)) \ .withColumn(dist, levenshtein(col(a.product_name), col(b.product_name))) \ .filter(col(dist) 2) \ .select(a.product_name, b.product_name, dist) \ .limit(100) # 仅取前100对避免笛卡尔爆炸实操心得levenshtein在PySpark中是O(n²)复杂度绝不能在全表上运行。我的技巧是先用sample(0.001)取1000行再做交叉比较准确率损失不到5%但耗时从小时级降到秒级。3.8 多列关联探查corr、crosstab与stat.approxQuantile的协同单列统计容易多列关系难。比如想看“用户年龄”和“购买金额”是否相关spark_df.corr(age, amount)只能给一个皮尔逊系数但无法告诉你在25-35岁区间是否强相关其他区间是否无关。我的分层探查法from pyspark.sql.functions import * # 步骤1用approxQuantile将age分5组五分位 age_quantiles spark_df.approxQuantile(age, [0.2, 0.4, 0.6, 0.8], 0.01) age_bins [0] age_quantiles [100] # 步骤2用bucketBy创建age_group列 from pyspark.sql.functions import when, col age_group_expr None for i, (low, high) in enumerate(zip(age_bins[:-1], age_bins[1:])): if i 0: age_group_expr when((col(age) low) (col(age) high), f{i1}_to_{i2}) else: age_group_expr age_group_expr.when((col(age) low) (col(age) high), f{i1}_to_{i2}) age_grouped spark_df.withColumn(age_group, age_group_expr) # 步骤3按age_group分组统计amount的分布 grouped_stats age_grouped.groupBy(age_group).agg( count(*).alias(count), avg(amount).alias(avg_amount), approxQuantile(amount, [0.25, 0.5, 0.75], 0.01).alias(amount_q25_q50_q75) )这样得到的不是单一相关系数而是可操作的业务洞察比如发现35-45岁组的avg_amount比其他组高40%且q7575分位达¥5000说明这个群体有高价值客户聚集值得单独建模。3.9 数据新鲜度验证max(event_time)与input_file_name()的双保险数据管道最怕“看起来在跑其实没更新”。光看max(event_time)不够因为时间字段可能被错误填充如全填成1970-01-01。我的双重验证法from pyspark.sql.functions import * # 方法1检查事件时间的最大值是否在合理窗口内 max_event_time spark_df.agg(max(event_time)).collect()[0][0] is_fresh (max_event_time and max_event_time date_sub(current_date(), 7)) # 近7天内 # 方法2检查数据来源文件的修改时间HDFS/S3 # 需启用spark.sql.files.ignoreMissingFilesfalse默认true file_info spark_df.select(input_file_name().alias(file_path)).distinct() # 解析S3路径中的last_modified时间需UDF此处略 # 或直接看文件名中的日期分区 partition_date spark_df.select( regexp_extract(input_file_name(), r/dt(\d{4}-\d{2}-\d{2})/, 1).alias(dt) ).filter(col(dt) ! ).agg(max(dt)).collect()[0][0] # 综合判断 freshness_report { max_event_time: max_event_time, latest_partition: partition_date, is_fresh_by_time: is_fresh, is_fresh_by_partition: (partition_date and partition_date date_sub(current_date(), 1).cast(string)) }某次在IoT设备数据中max_event_time显示最新是昨天但input_file_name()解析出的分区全是前天的追查发现是Kafka消费者组offset提交失败数据滞留在topic里没消费——这种问题单靠时间字段永远发现不了。3.10 跨表一致性探查exceptAll与intersectAll的精准比对数据治理常需验证主表与维度表的一致性比如orders.user_id是否都在users.user_id中存在。left_antijoin虽能实现但会丢失统计信息。我的方案是用集合运算# 主表orders的user_id去重 order_users orders.select(user_id).distinct() # 维度表users的user_id去重 user_users users.select(user_id).distinct() # 找出orders中有但users中没有的user_id孤儿记录 orphans order_users.exceptAll(user_users) orphan_count orphans.count() # 找出users中有但orders中从未出现的user_id沉默用户 silents user_users.exceptAll(order_users) silent_count silents.count() # 找出双方都有的user_id活跃用户 actives order_users.intersectAll(user_users) active_count actives.count() # 输出一致性报告 consistency_report { total_orders: orders.count(), total_users: users.count(), orphan_user_ids: orphan_count, silent_user_ids: silent_count, active_user_ids: active_count, orphan_ratio: orphan_count / orders.count() if orders.count() 0 else 0, silent_ratio: silent_count / users.count() if users.count() 0 else 0 }exceptAll和intersectAll是Spark 3.0引入的比旧版subtract和intersect更准确保留重复行语义且性能提升30%以上。某次在电商订单核验中发现orphan_ratio0.8%追查是CRM系统同步延迟导致及时拦截了错误的用户画像推送。3.11 自定义统计函数用pandas_udf实现分组百分位与移动平均PySpark内置函数有限复杂统计需UDF。但普通udf是逐行处理性能差。pandas_udf向量化UDF能批量处理效率提升10倍以上。例如计算每个品类的销售金额90分位from pyspark.sql.functions import pandas_udf from pyspark.sql.types import DoubleType import pandas as pd pandas_udf(returnTypeDoubleType()) def percentile_90(s: pd.Series) - float: return s.quantile(0.9) # 应用到分组聚合 category_percentiles orders.groupBy(category).agg( percentile_90(amount).alias(amount_p90) )再比如计算7日滚动销售额需按日期排序from pyspark.sql.window import Window # 先按date排序 window_spec Window.partitionBy(category).orderBy(date).rowsBetween(-6, 0) rolling_sum sum(amount).over(window_spec) category_rolling orders.withColumn(7d_rolling_sum, rolling_sum)注意pandas_udf要求Pandas版本与Spark兼容Spark 3.3推荐Pandas 1.3且内存配置要足够否则会OOM。我的经验是每个executor分配至少4GB内存给pandas_udf。3.12 EDA结果持久化从toPandas()到Delta Lake的演进早期我把所有统计结果collect()到Driver再toPandas()结果在100GB数据上Driver内存爆了。现在全部改用Delta Lake写入# 将统计结果作为Delta表写入 stats_summary.write \ .format(delta) \ .mode(overwrite) \ .option(mergeSchema, true) \ .save(/data/governance/eda_summary) # 后续可直接查询 spark.read.format(delta).load(/data/governance/eda_summary).show()Delta Lake的好处自动版本管理DESCRIBE HISTORY看每次EDA变更、ACID事务避免并发写冲突、Z-Order优化按table_name和run_date聚簇加速按表查询。某次客户要求回溯3个月前的数据质量报告我直接SELECT * FROM delta./data/governance/eda_summaryWHERE run_date 2023-10-015秒返回而旧方式要重跑整个EDA流程。4. 实操全流程以电商用户行为日志为例的端到端EDA实现4.1 场景设定与数据准备我们以某电商平台的user_behavior_log表为例该表每日增量约8亿行存储在HDFS上Schema如下字段名类型描述event_idstring事件唯一IDuser_idstring用户ID加密item_idstring商品IDcategorystring商品一级类目behaviorstring行为类型pv/click/cart/buyevent_timetimestamp事件发生时间pricedouble商品价格仅buy时有值provincestring用户省份数据已按dt日期分区组织当前需对dt2024-01-15分区做全量EDA。集群配置10个Worker节点每个16核64GB内存Spark 3.3.2。4.2 第一阶段Schema与元数据快速扫描10秒# 加载数据注意只读分区不触发全表扫描 log_df spark.read.table(user_behavior_log).filter(col(dt) 2024-01-15) # 步骤1基础Schema检查 print( Schema Overview ) log_df.printSchema() # 步骤2列级元数据统计使用前文analyze_schema函数 schema_analysis analyze_schema(log_df) schema_analysis.show(truncateFalse) # 步骤3分区信息验证 partition_info log_df.select( min(event_time).alias(min_event_time), max(event_time).alias(max_event_time), count(*).alias(total_rows), approx_count_distinct(user_id).alias(distinct_users), approx_count_distinct(item_id).alias(distinct_items) ).collect()[0] print(fPartition dt2024-01-15:) print(f Time Range: {partition_info[min_event_time]} to {partition_info[max_event_time]}) print(f Total Rows: {partition_info[total_rows]:,}) print(f Distinct Users: {partition_info[distinct_users]:,}) print(f Distinct Items: {partition_info[distinct_items]:,})实测结果total_rows 823,456,7898.2亿distinct_users 12,345,6781234万min_event_time 2024-01-15 00:00:02.123max_event_time 2024-01-15 23:59:59.999schema_analysis发现province列有1.2%的UNKNOWN值behavior列有0.03%的error值注意approx_count_distinct在8亿行上耗时仅4秒而countDistinct需18秒且内存压力大。这就是选择近似算法的价值。4.3 第二阶段核心列深度探查60秒4.3.1 用户行为分布category behavior# 行为类型分布用freqItems加速 behavior_freq log_df.freqItems([behavior], support0.001).select(explode(behavior_items).alias(behavior)).collect() behavior_df spark.createDataFrame(behavior_freq) # 类目分布先采样再freqItems sampled_log log_df.sample(False, 0.005) # 0.5%采样 category_freq sampled_log.freqItems([category], support0.01).select(explode(category_items).alias(category)).collect() category_df spark.createDataFrame(category_freq) # 合并统计 behavior_stats log_df.groupBy(behavior).count().withColumn( ratio, col(count) / lit(partition_info[total_rows]) ).orderBy(desc(count)) category_stats log_df.groupBy(category).count().withColumn( ratio, col(count) / lit(partition_info[total_rows]) ).orderBy(desc(count)) # 输出TOP10 print( Top 10 Behaviors ) behavior_stats.show(10, truncateFalse) print( Top 10 Categories ) category_stats.show(10, truncateFalse)关键发现behavior中pv占62.3%click占28.1%cart占7.2%buy仅2.4% —— 符合漏斗模型category中Electronics电子占31.5%Fashion