1. 项目概述为什么用 AWS EMR 做 PySpark 探索性分析而不是本地笔记本你手头有一份纽约出租车的完整历史票价数据集——近200GB的原始CSV包含2009到2016年超过17亿条行程记录字段包括上车/下车时间、经纬度、乘客数、支付方式、小费金额、总费用……光是用pandas读取一个分区就卡死pd.read_csv()报错“MemoryError”dask调度慢得像在等咖啡凉透而你连最基础的问题都还没问出口高峰期小费比例是否真的更高雨天打车是否更贵不同区域的平均等待时间差异有多大这就是我去年接手这个项目时的真实状态。不是理论不行是数据量已经越过了单机处理的物理边界。PySpark不是“另一个Python库”它是把计算逻辑编译成DAG有向无环图再由集群分发到多台机器并行执行的引擎——它不把数据全加载进内存而是按需拉取、流式处理、惰性求值。但关键问题来了你不会为了跑一次EDA就去采购三台r5.4xlarge服务器、配YARN、调Shuffle分区、修HDFS权限吧这就是AWS EMR的价值所在它不是“云上的Hadoop”而是“开箱即用的PySpark生产环境”。你点几下控制台或写三行CLI命令10分钟内就能拿到一个预装Spark 3.4、Python 3.9、Delta Lake 2.4、甚至带JupyterHub的完整集群所有底层依赖如Snappy压缩、Parquet向量化读取、Kerberos认证都已调优完毕。我实测过同样一份NYC Taxi数据在本地Mac M1上用pandas做基础统计要47分钟在EMR r5.2xlarge4核16GB单节点模式下PySpark SQL完成相同聚合仅需82秒扩展到3个core节点后耗时压到29秒——而且全程不用改一行代码只改了spark-submit --master yarn里的资源配置。这不是“云原生”的营销话术是真实可量化的工程效率跃迁。本文不讲抽象概念只带你从零部署EMR集群、把17亿行原始CSV转成高效Parquet、用PySpark DataFrame API完成典型EDA流程缺失值热力图、地理空间分布、时间序列趋势、异常值检测最后把结果可视化回传到本地Jupyter。所有步骤我都录了操作日志参数值、报错截图、耗时对比表全部附在文末。适合刚学完Spark RDD但没碰过真实集群的新手也适合想验证云上EDA可行性的数据工程师。2. 整体设计与技术选型逻辑为什么是EMRPySpark而不是Glue、SageMaker或本地Spark2.1 集群平台选型EMR vs Glue vs SageMaker Notebook很多人看到“云上PySpark”第一反应是AWS Glue——毕竟它标榜“无服务器”。但实际用过就知道Glue的JobBook是为ETL流水线设计的它的PySpark运行时基于Spark 2.x截至2024年Q2仍不支持Spark 3.3的AQE动态优化且强制使用--enable-glue-datacatalog导致你无法自由指定Hive Metastore或直接读写S3路径。更致命的是资源粒度Glue最小分配单位是DPUData Processing Unit1 DPU ≈ 4GB内存1 vCPU但实际内存可用率只有60%意味着你申请4 DPU16GB却只能用9.6GB而NYC Taxi单个分区文件就超2GB很容易OOM。我试过Glue 4.0版本跑df.describe()直接被Kill。SageMaker Notebook呢它本质是EC2实例JupyterLab虽然能装最新版PySpark但它没有YARN资源管理器。所有Spark任务都在单个Notebook实例上运行相当于把集群退化成单机——你买个c5.4xlarge16核64GB确实比Mac强但无法横向扩展也无法利用EMR的Spot Instance竞价机制省70%成本。更重要的是SageMaker的Spark配置需要手动写spark-defaults.conf连spark.sql.adaptive.enabled这种关键开关都要自己加而EMR在创建集群时就提供勾选框。EMR的优势恰恰卡在中间它既不像Glue那样阉割功能也不像SageMaker那样放弃集群能力。它的核心价值在于预集成可定制成本可控。比如EMR 6.15默认启用Spark 3.4.1 AQE Dynamic Partition Pruning而你要在SageMaker上手动编译Spark源码才能获得同等特性。再比如Spot Instance支持——我用--instance-fleet参数定义core节点组时设置BidPriceAsPercentageOfOnDemandPrice70实际运行中85%的core节点都是Spot实例月均成本从$1,200降到$180。这不是玄学是AWS官方文档明确写的竞价策略。提示EMR的“Managed Scaling”功能在EDA场景反而要关掉。因为EDA是短时爆发型负载可能连续提交10个SQL查询然后空闲20分钟Managed Scaling会误判为低负载而缩容导致正在运行的任务失败。我的做法是在创建集群时禁用自动扩缩用aws emr modify-instance-groups命令手动调整节点数。2.2 数据格式选型为什么坚持用Parquet而不是CSV或JSON原始NYC Taxi数据是CSV格式共22个字段单文件最大达12GB。如果直接用spark.read.csv(s3://bucket/raw/yellow_tripdata_2015-01.csv)会发生三件可怕的事第一Schema推断失效CSV没有类型定义Spark必须扫描全文件才能猜出passenger_count是整数还是字符串而12GB文件扫描要11分钟第二I/O放大严重CSV是纯文本每行都要解析分隔符、转义字符、引号CPU时间全耗在字符串处理上第三无法列裁剪你想查pickup_datetime和fare_amountSpark仍得把整行22列全读进内存再过滤。Parquet完美解决这三点。它是列式存储自描述Schema内置Snappy压缩。我用df.write.mode(overwrite).parquet(s3://bucket/parquet/yellow_2015/)把2015年数据转成Parquet后体积从12GB压缩到1.8GB压缩率6.7倍更重要的是——Schema自动保存在_metadata文件里下次读取无需推断查询只读取涉及的列df.select(pickup_datetime, fare_amount).show()实际I/O量下降83%支持谓词下推Predicate Pushdowndf.filter(fare_amount 100).count()会在S3端就过滤掉99%的行避免网络传输。有人问“那Delta Lake呢”Delta是Parquet的超集增加ACID事务和时间旅行但EDA阶段不需要这些。我实测过纯Parquet读取速度比Delta快1.3倍因Delta要校验事务日志而Delta的VACUUM命令在EDA临时分析中毫无意义。所以我的原则是EDA用Parquet生产ETL用Delta。2.3 开发环境选型为什么用EMR自带的JupyterHub而不是本地VS Code Remote本地VS Code通过SSH连接EMR Master节点确实可行但存在两个硬伤文件同步瓶颈你改一行Python代码VS Code要通过SCP上传到EMR而EMR的EBS卷I/O性能一般10KB脚本上传常卡2秒依赖隔离困难本地装的pyspark3.4.1和EMR集群的pyspark3.4.1看似一样但EMR的PySpark包里嵌入了AWS SDK 2.x和S3A FileSystem适配器本地pip install的版本没有这些导致s3a://路径报ClassNotFoundException。EMR自带的JupyterHub通过http://master-ip:8443访问则天然解决这些问题Notebook直接运行在Master节点所有Spark上下文、Hadoop配置、S3凭证都已注入支持%%spark魔法命令可直接写SQL查询结果自动渲染成表格文件存于EMR的EBS卷读写延迟低于10ms比S3快三个数量级。当然它也有缺点UI老旧、不支持VS Code的调试器。我的折中方案是——开发用JupyterHub写逻辑调试用本地PyCharm远程attach到EMR的Py4J网关。具体操作见第3.4节。3. 核心细节解析与实操要点从集群创建到数据探查的避坑指南3.1 EMR集群创建避开5个致命配置错误创建EMR集群看似简单但90%的初学者会在以下环节翻车。我用aws emr create-clusterCLI命令为例比控制台更易复现逐条说明正确配置aws emr create-cluster \ --name nyc-taxi-eda \ --release-label emr-6.15.0 \ --applications NameSpark NameHive \ --ec2-attributes KeyNamemy-key-pair,InstanceProfileEMR_EC2_DefaultRole \ --service-role EMR_DefaultRole \ --instance-fleet file://fleet-config.json \ --configurations file://spark-config.json \ --bootstrap-actions Paths3://my-bucket/bootstrap.sh \ --log-uri s3://my-bucket/emr-logs/ \ --auto-terminate \ --region us-east-1错误1用--instance-type而非--instance-fleet--instance-type只能建固定规格集群无法混合On-Demand和Spot实例。而--instance-fleet允许你定义多个实例组比如{ InstanceFleetType: CORE, TargetOnDemandCapacity: 1, TargetSpotCapacity: 2, InstanceTypes: [ {InstanceType: r5.2xlarge}, {InstanceType: r5.4xlarge} ] }这样既能保证1台稳定On-Demand节点又能用2台低价Spot节点扛计算压力成本直降55%。错误2忽略spark-config.json中的内存调优EMR默认的spark.executor.memory是1GB这对NYC Taxi数据完全不够。我在spark-config.json里强制覆盖[ { Classification: spark-defaults, Properties: { spark.executor.memory: 8g, spark.driver.memory: 4g, spark.sql.adaptive.enabled: true, spark.sql.adaptive.coalescePartitions.enabled: true } } ]注意spark.executor.memory不能超过节点内存的80%r5.2xlarge有64GB所以设8g安全否则YARN会Kill Executor。错误3Bootstrap脚本未安装必要Python包EMR自带的PySpark不包含pandas、matplotlib等可视化库。我的bootstrap.sh内容如下#!/bin/bash sudo /usr/bin/python3 -m pip install --upgrade pip sudo /usr/bin/python3 -m pip install pandas matplotlib seaborn plotly # 关键修复PySpark与pandas的兼容性 sudo /usr/bin/python3 -m pip install pyarrow11.0.0这里pyarrow11.0.0是重点——Spark 3.4.1要求Arrow 11.x装12.x会报ArrowInvalid: Unsupported type。错误4Log URI指向不存在的S3桶--log-uri s3://my-bucket/emr-logs/中的my-bucket必须提前创建且EMR角色要有s3:PutObject权限。否则集群创建成功但日志无法写入排查问题时两眼抓瞎。错误5忘记--auto-terminateEDA是临时任务不设此参数集群会一直运行按小时计费。我习惯加--auto-terminate任务结束自动销毁配合CloudWatch告警监控费用。注意创建后务必记下Master节点的Public DNS如ec2-12-34-56-78.compute-1.amazonaws.com这是后续SSH和Jupyter访问的入口。用aws emr describe-cluster --cluster-id j-XXXXXXXXXX --query Cluster.MasterPublicDnsName可快速获取。3.2 NYC Taxi数据预处理从原始CSV到高效Parquet的4步转换原始数据来自 NYC TLC官网 下载后是ZIP包解压得CSV。但直接上传S3会遇到编码问题——2015年后的CSV用UTF-8而2009-2014年用Latin-1。我的处理流程如下Step 1S3目录结构设计不把所有CSV扔进一个文件夹按年份分区便于后续partitionBys3://my-bucket/raw/yellow/2009/ s3://my-bucket/raw/yellow/2010/ ... s3://my-bucket/raw/yellow/2016/每个年份文件夹下放对应年份的CSV如yellow_tripdata_2015-01.csv。这样Spark读取时能自动识别分区列。Step 2编写PySpark清洗脚本ingest.pyfrom pyspark.sql import SparkSession from pyspark.sql.functions import col, to_timestamp, year, month, when, isnan, isnull from pyspark.sql.types import * spark SparkSession.builder \ .appName(NYC-Taxi-Ingest) \ .getOrCreate() # 定义Schema避免推断关键 schema StructType([ StructField(VendorID, IntegerType(), True), StructField(tpep_pickup_datetime, StringType(), True), # 先读String再转Timestamp StructField(tpep_dropoff_datetime, StringType(), True), StructField(passenger_count, IntegerType(), True), StructField(trip_distance, DoubleType(), True), StructField(RatecodeID, IntegerType(), True), StructField(store_and_fwd_flag, StringType(), True), StructField(PULocationID, IntegerType(), True), StructField(DOLocationID, IntegerType(), True), StructField(payment_type, IntegerType(), True), StructField(fare_amount, DoubleType(), True), StructField(extra, DoubleType(), True), StructField(mta_tax, DoubleType(), True), StructField(tip_amount, DoubleType(), True), StructField(tolls_amount, DoubleType(), True), StructField(improvement_surcharge, DoubleType(), True), StructField(total_amount, DoubleType(), True), StructField(congestion_surcharge, DoubleType(), True) ]) # 读取2015年所有CSV自动glob匹配 df spark.read \ .option(header, true) \ .option(inferSchema, false) \ .schema(schema) \ .csv(s3://my-bucket/raw/yellow/2015/*.csv) # 清洗转时间戳、处理空值、过滤异常值 df_clean df \ .withColumn(pickup_datetime, to_timestamp(col(tpep_pickup_datetime))) \ .withColumn(dropoff_datetime, to_timestamp(col(tpep_dropoff_datetime))) \ .filter(col(pickup_datetime).isNotNull()) \ .filter(col(dropoff_datetime).isNotNull()) \ .filter(col(trip_distance) 0) \ .filter(col(fare_amount) 0) \ .filter(col(passenger_count).between(1, 6)) \ .drop(tpep_pickup_datetime, tpep_dropoff_datetime) # 删除原始字符串列 # 写入Parquet按年月分区提升后续查询效率 df_clean \ .withColumn(year, year(col(pickup_datetime))) \ .withColumn(month, month(col(pickup_datetime))) \ .write \ .mode(overwrite) \ .partitionBy(year, month) \ .parquet(s3://my-bucket/parquet/yellow_2015/)Step 3提交作业到EMRspark-submit \ --master yarn \ --deploy-mode client \ --conf spark.sql.adaptive.enabledtrue \ ingest.py注意--deploy-mode clientDriver运行在Master节点便于查看日志若用cluster模式Driver在YARN容器里日志分散难追踪。Step 4验证Parquet质量在JupyterHub里运行df spark.read.parquet(s3://my-bucket/parquet/yellow_2015/) print(f行数: {df.count()}) print(f分区数: {len(df.inputFiles())}) df.printSchema()理想输出行数≈1.2亿分区数≈12每月一个分区Schema显示pickup_datetime为timestamp类型。若count()卡住大概率是S3权限没配好。实操心得第一次运行ingest.py时我忘了--conf spark.sql.adaptive.enabledtrue结果Shuffle阶段生成了2000个task耗时42分钟。加上AQE后task数自动合并为128个耗时压到6.8分钟。AQE不是银弹但它让新手免于手动调spark.sql.shuffle.partitions。3.3 EDA核心分析用PySpark实现5类典型探索3.3.1 缺失值分析不只是df.na.drop()那么简单pandas的df.isnull().sum()在PySpark里对应df.agg(*[count(when(isnull(c) | isnan(c), c)).alias(c) for c in df.columns])但这只能给总数。EDA需要知道缺失模式是整列缺失如2009年没congestion_surcharge还是随机缺失如tip_amount在现金支付时为空我的方案是用crosstab看关联性# 检查payment_type和tip_amount的关系 df.groupBy(payment_type) \ .agg( count(*).alias(total), count(when(col(tip_amount).isNull(), 1)).alias(tip_null_count) ) \ .withColumn(null_ratio, col(tip_null_count) / col(total)) \ .orderBy(null_ratio) \ .show()输出显示payment_type2Cash时tip_null_count占比99.7%而payment_type1Credit Card时仅0.3%。这说明现金支付几乎不记录小费——不是数据损坏是业务逻辑。因此后续分析小费时应过滤payment_type 1。注意isnan()只对数值列有效isNull()对所有类型有效。混用二者才能覆盖全场景。3.3.2 地理空间分布用经纬度画热力图原始数据有pickup_longitude、pickup_latitude但直接df.select(pickup_longitude, pickup_latitude).toPandas()会OOM。正确做法是采样聚合# 先采样100万行保证地理分布代表性 sample_df df.sample(0.001).limit(1000000) # 0.001抽样率再限100万 # 转为GeoHash精度5位约5km×5km网格 from pyspark.sql.functions import udf from pyspark.sql.types import StringType import geohash2 geohash_udf udf(lambda lon, lat: geohash2.encode(lat, lon, precision5), StringType()) geo_df sample_df.withColumn(geohash, geohash_udf(pickup_longitude, pickup_latitude)) # 按GeoHash聚合计数 geo_counts geo_df.groupBy(geohash).count().orderBy(count, ascendingFalse) geo_counts.show(10)结果前10名GeoHash全集中在曼哈顿中城如dr5ru对应时代广场验证了数据合理性。再把geo_counts转成Pandas用folium画热力图import folium from folium.plugins import HeatMap pandas_df geo_counts.toPandas() # 将GeoHash转回经纬度中心点 pandas_df[lat] pandas_df[geohash].apply(lambda g: geohash2.decode(g)[0]) pandas_df[lon] pandas_df[geohash].apply(lambda g: geohash2.decode(g)[1]) m folium.Map(location[40.7128, -74.0060], zoom_start11) HeatMap(datapandas_df[[lat, lon, count]].values, radius15).add_to(m) m.save(nyc_pickup_heatmap.html)3.3.3 时间序列趋势提取小时、星期、节假日特征pickup_datetime是时间分析核心。PySpark的date_format函数比pandas的dt.hour更高效from pyspark.sql.functions import date_format, dayofweek, hour, weekofyear, lit trend_df df \ .withColumn(hour, hour(pickup_datetime)) \ .withColumn(day_of_week, dayofweek(pickup_datetime)) \ # 1Sunday, 2Monday... .withColumn(week_of_year, weekofyear(pickup_datetime)) \ .withColumn(is_weekend, when(col(day_of_week).isinCollection([1, 7]), 1).otherwise(0)) # 计算每小时订单量验证早高峰 hourly_count trend_df.groupBy(hour).count().orderBy(hour) hourly_count.show()输出显示hour17下午5点和hour18下午6点订单量峰值符合通勤规律。但注意dayofweek返回1-7而业务常需1Monday所以实际用when(col(day_of_week)1, 7).otherwise(col(day_of_week)-1)重映射。3.3.4 异常值检测用IQR法替代简单filter(fare_amount 1000)fare_amount有极端值如$10,000但直接filter会误杀。IQR四分位距更科学from pyspark.sql.functions import expr, percentile_approx # 计算Q1、Q3用approx_percentile避免全排序 q1_q3 df.agg( percentile_approx(fare_amount, 0.25).alias(q1), percentile_approx(fare_amount, 0.75).alias(q3) ).collect()[0] iqr q1_q3[q3] - q1_q3[q1] lower_bound q1_q3[q1] - 1.5 * iqr upper_bound q1_q3[q3] 1.5 * iqr print(fIQR范围: [{lower_bound:.2f}, {upper_bound:.2f}]) # 输出: IQR范围: [2.50, 42.00] # 过滤异常值 df_filtered df.filter((col(fare_amount) lower_bound) (col(fare_amount) upper_bound))percentile_approx用TDigest算法内存占用仅为精确百分位的1/10017亿行数据计算Q1/Q3仅耗时23秒。3.3.5 相关性分析用corr()和crosstab组合解读df.stat.corr(fare_amount, tip_amount)返回0.62说明正相关。但相关系数掩盖了非线性关系。我用crosstab看分段# 按fare_amount分5档 from pyspark.sql.functions import when, col df_tipped df.filter(col(tip_amount) 0) df_tipped_grouped df_tipped \ .withColumn(fare_bin, when(col(fare_amount) 10, 0-10) .when(col(fare_amount) 20, 10-20) .when(col(fare_amount) 30, 20-30) .when(col(fare_amount) 50, 30-50) .otherwise(50)) # 计算每档小费比例均值 df_tipped_grouped \ .groupBy(fare_bin) \ .agg( avg(tip_amount).alias(avg_tip), avg(tip_amount) / avg(fare_amount) * 100).alias(tip_rate_pct) ) \ .orderBy(fare_bin) \ .show()结果揭示真相0-10档小费比例均值22.3%50档降至12.1%——便宜的短途单反而小费比例更高。这比单个相关系数更有业务洞察力。4. 实操过程与核心环节实现从零开始的完整工作流4.1 环境准备SSH连接、Jupyter配置与S3权限验证创建集群后第一步是建立安全连接。不要用密码登录用密钥对# 下载密钥对假设叫emr-key.pem chmod 400 emr-key.pem ssh -i emr-key.pem hadoopec2-12-34-56-78.compute-1.amazonaws.com登录后验证Sparkpyspark --version # 应输出3.4.1 spark-shell --version # 同样验证Scala版接着启动JupyterHubEMR 6.15默认已安装sudo systemctl start jupyterhub sudo systemctl enable jupyterhub访问https://ec2-12-34-56-78.compute-1.amazonaws.com:8443用hadoop用户登录密码在EMR创建时设置。S3权限验证是生死线。在Jupyter里运行# 测试S3读写 spark.sparkContext.textFile(s3://my-bucket/test.txt).count() # 应返回行数 spark.range(10).write.mode(overwrite).parquet(s3://my-bucket/test-parquet/)若报错AccessDeniedException检查两点EMR EC2 Role是否附加了AmazonS3FullAccess策略测试阶段可开生产需最小权限S3桶策略是否允许Principal: {Service: ec2.amazonaws.com}EMR用EC2角色访问S3。4.2 数据加载与Schema诊断如何一眼看出CSV脏在哪加载原始CSV时永远加.option(mode, DROPMALFORMED)df_raw spark.read \ .option(header, true) \ .option(mode, DROPMALFORMED) \ # 跳过格式错误行 .option(inferSchema, true) \ .csv(s3://my-bucket/raw/yellow/2015-01.csv)DROPMALFORMED会丢弃1,2015-01-01 00:00:00,2015-01-01 00:12:00,1,1.2,这种末尾缺字段的坏行。但更关键的是看df_raw.printSchema()的输出root |-- VendorID: string (nullable true) |-- tpep_pickup_datetime: string (nullable true) |-- tpep_dropoff_datetime: string (nullable true) |-- passenger_count: integer (nullable true) |-- trip_distance: double (nullable true) |-- RatecodeID: string (nullable true) ← 问题RatecodeID应是int却是stringRatecodeID列为string说明该列有非数字字符如N/A或空格。此时不能盲目cast(int)先查异常值df_raw.filter(col(RatecodeID).rlike([^0-9])).select(RatecodeID).distinct().show() # 输出: |N/A|, | |, |99|于是清洗逻辑变成df_clean df_raw \ .withColumn(RatecodeID, when(col(RatecodeID).isinCollection([N/A, , 99]), None) .otherwise(col(RatecodeID).cast(int)))4.3 EDA全流程代码可直接复制粘贴的Jupyter Notebook以下是我在JupyterHub中实际运行的完整EDA Notebook已删减注释保留核心逻辑# Cell 1: 初始化 from pyspark.sql import SparkSession from pyspark.sql.functions import * from pyspark.sql.types import * import matplotlib.pyplot as plt import seaborn as sns spark SparkSession.builder \ .appName(NYC-Taxi-EDA) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() # Cell 2: 加载Parquet数据已清洗 df spark.read.parquet(s3://my-bucket/parquet/yellow_2015/) # Cell 3: 基础统计替代pandas describe df.select( count(*).alias(total_rows), count(when(col(fare_amount) 0, 1)).alias(neg_fare), count(when(col(trip_distance) 0, 1)).alias(zero_distance), count(when(col(passenger_count) 0, 1)).alias(zero_passenger) ).show() # Cell 4: 时间分布直方图用PySpark聚合避免toPandas hourly_dist df.groupBy(hour(pickup_datetime).alias(hour)) \ .count() \ .orderBy(hour) \ .toPandas() plt.figure(figsize(10,4)) sns.barplot(datahourly_dist, xhour, ycount) plt.title(Hourly Trip Count in 2015) plt.savefig(/home/hadoop/hourly_dist.png) # 保存到EMR本地注意plt.savefig()路径必须是EMR节点的本地路径如/home/hadoop/不能是S3。之后用scp下载到本地scp -i emr-key.pem hadoopec2-...:/home/hadoop/hourly_dist.png ./local/4.4 本地调试技巧PyCharm远程Attach到EMR的Py4J网关当Jupyter里逻辑复杂需断点调试时PyCharm比Jupyter强大得多。步骤如下在EMR Master节点启动Py4J网关# 修改spark-defaults.conf echo spark.driver.extraJavaOptions -agentlib:jdwptransportdt_socket,servery,suspendn,address*:5005 /etc/spark/conf/spark-defaults.conf sudo systemctl restart spark-history-serverPyCharm中配置Remote JVM DebugHost填EMR Master公网IPPort填5005在PyCharm中写代码spark SparkSession.builder.master(yarn).getOrCreate()点击Debug按钮PyCharm会连接到EMR的JVM所有断点生效。此方法让我定位到一个隐藏Bugdf.filter(tip_amount 0).count()返回0但df.filter(col(tip_amount) 0).count()返回正确值。原因是SQL字符串解析器把tip_amount当列名而col()函数明确指向DataFrame列对象。5. 常见问题与排查技巧实录踩过的12个坑及解决方案5.1 经典报错速查表报错信息根本原因解决方案复现概率java.lang.OutOfMemoryError: GC overhead limit exceededExecutor内存不足GC时间超98%增大spark.executor.memory或减小spark.sql.files.maxPartitionBytes默认128MB设为64MB★★★★★org.apache.hadoop.fs.s3a.AWSBadRequestException: Status Code: 400S3A FileSystem版本不匹配在spark-defaults.conf中添加spark.hadoop.fs.s3a.impl org.apache.hadoop.fs.s3a.S3AFileSystem★★★★☆Py4JJavaError: An error occurred while calling o53.sql. : java.lang.ClassNotFoundException: org.apache.spark.sql.hive.HiveSessionStateBuilderHive支持未启用创建集群时加--applications NameHive或在SparkSession中加.enableHiveSupport()★★★☆☆pyspark.sql.utils.AnalysisException: The format of the given file is not supported.Parquet文件损坏或Schema不一致用parquet-tools检查parquet-tools meta s3://bucket/part-00000-xxx.snappy.parquet★★☆☆☆5.2 性能调优实战从32分钟到4.2分钟的5次迭代我用df.agg(avg(fare_amount)).show()作为基准测试记录每次优化后的耗时| 迭代 | 操作 |