AWS EMR+PySpark大规模EDA实战：从17亿行出租车数据说起

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.2xlarge（4核16GB）单节点模式下，PySpark SQL完成相同聚合仅需82秒；扩展到3个core节点后，耗时压到29秒——而且全程不用改一行代码，只改了spark-submit --master yarn里的资源配置。这不是“云原生”的营销话术，是真实可量化的工程效率跃迁。本文不讲抽象概念，只带你从零部署EMR集群、把17亿行原始CSV转成高效Parquet、用PySpark DataFrame API完成典型EDA流程（缺失值热力图、地理空间分布、时间序列趋势、异常值检测），最后把结果可视化回传到本地Jupyter。所有步骤我都录了操作日志，参数值、报错截图、耗时对比表全部附在文末。适合刚学完Spark RDD但没碰过真实集群的新手，也适合想验证云上EDA可行性的数据工程师。

2. 整体设计与技术选型逻辑：为什么是EMR+PySpark，而不是Glue、SageMaker或本地Spark？

2.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最小分配单位是DPU（Data Processing Unit），1 DPU ≈ 4GB内存+1 vCPU，但实际内存可用率只有60%，意味着你申请4 DPU（16GB）却只能用9.6GB，而NYC Taxi单个分区文件就超2GB，很容易OOM。我试过Glue 4.0版本跑df.describe()，直接被Kill。

SageMaker Notebook呢？它本质是EC2实例+JupyterLab，虽然能装最新版PySpark，但它没有YARN资源管理器。所有Spark任务都在单个Notebook实例上运行，相当于把集群退化成单机——你买个c5.4xlarge（16核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节点组时，设置BidPriceAsPercentageOfOnDemandPrice=70，实际运行中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_amount，Spark仍得把整行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 Pushdown），df.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秒；
• 依赖隔离困难：本地装的pyspark==3.4.1和EMR集群的pyspark==3.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 Name=Spark Name=Hive \ --ec2-attributes KeyName=my-key-pair,InstanceProfile=EMR_EC2_DefaultRole \ --service-role EMR_DefaultRole \ --instance-fleet file://fleet-config.json \ --configurations file://spark-config.json \ --bootstrap-actions Path=s3://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。

错误3：Bootstrap脚本未安装必要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 pyarrow==11.0.0

这里pyarrow==11.0.0是重点——Spark 3.4.1要求Arrow 11.x，装12.x会报ArrowInvalid: Unsupported type。

错误4：Log URI指向不存在的S3桶
--log-uri s3://my-bucket/emr-logs/中的my-bucket必须提前创建，且EMR角色要有s3:PutObject权限。否则集群创建成功但日志无法写入，排查问题时两眼抓瞎。

错误5：忘记--auto-terminate
EDA是临时任务，不设此参数集群会一直运行，按小时计费。我习惯加--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 1：S3目录结构设计
不把所有CSV扔进一个文件夹！按年份分区，便于后续partitionBy：

s3://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.py）

from 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：提交作业到EMR

spark-submit \ --master yarn \ --deploy-mode client \ --conf spark.sql.adaptive.enabled=true \ ingest.py

注意--deploy-mode client：Driver运行在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.enabled=true，结果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_type=2（Cash）时tip_null_count占比99.7%，而payment_type=1（Credit 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, precision=5), StringType()) geo_df = sample_df.withColumn("geohash", geohash_udf("pickup_longitude", "pickup_latitude")) # 按GeoHash聚合计数 geo_counts = geo_df.groupBy("geohash").count().orderBy("count", ascending=False) 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_start=11) HeatMap(data=pandas_df[["lat", "lon", "count"]].values, radius=15).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")) \ # 1=Sunday, 2=Monday... .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()

输出显示：hour=17（下午5点）和hour=18（下午6点）订单量峰值，符合通勤规律。但注意：dayofweek返回1-7，而业务常需1=Monday，所以实际用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(f"IQR范围: [{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/100，17亿行数据计算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" [e-mail]

登录后验证Spark：

pyspark --version # 应输出3.4.1 spark-shell --version # 同样验证Scala版

接着启动JupyterHub（EMR 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，检查两点：

1. EMR EC2 Role是否附加了AmazonS3FullAccess策略（测试阶段可开，生产需最小权限）；
2. 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，却是string

RatecodeID列为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(data=hourly_dist, x="hour", y="count") 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 hadoop@ec2-...:/home/hadoop/hourly_dist.png ./local/

4.4 本地调试技巧：PyCharm远程Attach到EMR的Py4J网关

当Jupyter里逻辑复杂需断点调试时，PyCharm比Jupyter强大得多。步骤如下：

1. 在EMR Master节点启动Py4J网关：

   # 修改spark-defaults.conf echo "spark.driver.extraJavaOptions -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005" >> /etc/spark/conf/spark-defaults.conf sudo systemctl restart spark-history-server

2. PyCharm中配置Remote JVM Debug，Host填EMR Master公网IP，Port填5005；
3. 在PyCharm中写代码，spark = SparkSession.builder.master("yarn").getOrCreate()；
4. 点击Debug按钮，PyCharm会连接到EMR的JVM，所有断点生效。

此方法让我定位到一个隐藏Bug：df.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 exceeded	Executor内存不足，GC时间超98%	增大spark.executor.memory，或减小spark.sql.files.maxPartitionBytes（默认128MB，设为64MB）	★★★★★
org.apache.hadoop.fs.s3a.AWSBadRequestException: Status Code: 400	S3A 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.HiveSessionStateBuilder	Hive支持未启用	创建集群时加--applications Name=Hive，或在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()作为基准测试，记录每次优化后的耗时：

| 迭代 | 操作 |