200GB出租车数据单机分析：PySpark+Dask+Datashader多引擎实践

简介：针对NYC出租车数据集（约200 GB）的处理与分析教学包，面向使用Python生态处理大规模数据的开发者，覆盖从AWS EC2 Hadoop集群部署、Cloudera平台搭建，到借助Datashader完成可视化的工作流程。包内共13个文件，其中5个.q为Hive建表与统计查询脚本，1个ipynb为可视化分析Notebook，另有sh/py脚本用于数据集下载，png为可视化结果图，md为说明文档，整体压缩包仅1.68MB。已有1383人学习下载。资料通过greenTaxi_visualization等案例展示无Datashader与有Datashader的差异，并提供建表、记录统计、绘图等完整脚本，可直接参考复现200 GB级数据的分析流程，适合想了解Hadoop+Python数据可视化实践的进阶学习者。

1. 200 GB出租车数据怎么分析：一个Jupyter Notebook驱动的多引擎方案

第一次拿到这份纽约出租车公开数据集时，我在Jupyter Notebook里想都没想就执行了pd.read_csv('yellow_tripdata_2022-01.csv')，结果进度条卡在三分钟的位置，内存从16GB一路冲到90%，内核直接重启。这不是我的电脑太弱，而是这套数据总共有超过200GB，单机内存根本没有一次吞下的可能。这个项目的意义在于：它没有让你先部署一套Hadoop集群，而是用PySpark、Dask、Datashader三件套，在这台个人电脑上把200GB数据完成了从清洗、统计到可视化的全流程。适合想处理大数据但又不打算养集群的开发者，也适合想从Pandas迁移到Spark技术栈的老手。

2. 架构与选型：PySpark、Dask、Datashader怎么分工

2.1 为什么这个项目选了三套引擎而不是一套

按理说PySpark自己就能做ETL、统计和一部分可视化准备，Dask也能处理200GB数据，为什么项目里同时用三套？这是我拆完整个仓库之后觉得最有意思的地方。实际处理链路是这样的：原始CSV是文本格式，解析开销大，用PySpark做一次性清洗，转成列式存储Parquet；转完之后，日常探索用Dask读Parquet，因为Dask的DataFrame API和Pandas几乎一模一样，写groupby、聚合不需要火星语法；最后画图用Datashader，因为PySpark和Dask都不擅长渲染上亿个坐标点，它们只能聚合出统计结果，画点阵图是Datashader的看家本领。

这个分工本质上是按计算阶段选工具：ETL阶段要的是吞吐量，探索阶段要的是低延迟，出图阶段要的是像素级聚合。三个库在各自阶段都有别人替代不了的优势，硬塞给其中任何一个都会很难受。我刚开始想过只留PySpark，但画图部分写出来丑得没法看，等值线和热力图的效果完全不是同一个量级。

2.2 项目目录设计与数据流走向

仓库结构是典型的notebook驱动分析工程，不是那种堆一堆.py的框架项目：

nyc-taxi-analysis/ ├── data/ │ ├── raw/ # 原始CSV，按月份分片 │ └── parquet/ # ETL输出列式存储 ├── notebooks/ │ ├── 01_etl_with_pyspark.ipynb │ ├── 02_stats_with_dask.ipynb │ └── 03_visualize_with_datashader.ipynb ├── scripts/ │ ├── etl.py # CLI版ETL脚本 │ └── make_viz.py # 批量出图脚本 └── requirements.txt

data/raw下放的是从官方页面下载的月度CSV，每个文件少则几百MB，多则接近2GB；data/parquet是ETL之后生成的列式文件，按年月分区，查询时可以直接跳过无关分区。notebooks目录按数字编号，从01到03，对应完整分析链路的三个步骤，每个notebook只保留当前阶段的代码，不把ETL和可视化揉在一个文件里。scripts里的两个.py是notebook的脚本化版本，跑批或定时任务时用，避免每次都要人工打开Jupyter Notebook点运行。我一般会把核心逻辑先写在notebook里验证，确认无误后再抽成脚本，这也是二次开发时最常见的需求。

2.3 环境准备：一台16GB内存机器就够了

这个项目我是在一台16GB内存、4核CPU的笔记本上跑通的。requirements.txt长这样：

pyspark==3.4.1 dask[complete]==2023.5.0 datashader==0.14.4 pyarrow==12.0.1 jupyterlab==4.0.0

安装直接用pip就够了，不需要单独装Java，因为PySpark 3.4以后自带Spark本地运行所需的Java运行时。安装完成后，我会额外设置两个环境变量：

export PYSPARK_PYTHON=python3 export PYSPARK_DRIVER_PYTHON=jupyter

第一个变量是让Spark worker统一使用系统python3，避免worker和driver解释器不一致；第二个严格来说不是必须，但如果你希望在Jupyter Notebook里直接跑spark.read.csv，提前把它导进去会让session初始化更干净。这两行我每次新建分析环境都会先写上，已经成了条件反射。

提示：PySpark自带的Java运行时只用于本地执行，不要把它当成生产环境的集群配置。

2.4 单机跑Spark的内存参数怎么调

从pandas切换到PySpark，最常见的问题是“为什么我的Spark比pandas还慢”。因为Spark默认配置是为集群设计的，shuffle默认200个分区，单机上会直接卡死。我通常会这样设置：

from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("nyc-taxi-etl") \ .config("spark.sql.shuffle.partitions", 8) \ .config("spark.executor.memory", "8g") \ .config("spark.driver.memory", "6g") \ .config("spark.ui.showConsoleProgress", "true") \ .getOrCreate()

参数说明：

这段配置直接决定了ETL能不能顺利跑完。spark.sql.shuffle.partitions是最容易让你翻车的参数，200是集群默认值，在单机4核上会产生200个小文件，导致每个task分配不到一个核，反而把磁盘IO拉满。调成8之后，整个ETL从半小时缩到十分钟，性能差距大到像玄学。

3. 核心模块拆解：从两亿条CSV到Parquet的完整链路

3.1 PySpark ETL：CSV变Parquet

打开notebooks/01_etl_with_pyspark.ipynb，第一段代码是建立SparkSession，第二段开始读数据。这个demo里数据量大约是两亿条记录，散落在多个月度CSV中：

from pyspark.sql import SparkSession from pyspark.sql.functions import col, to_timestamp, year, month spark = SparkSession.builder \ .appName("nyc-taxi-etl") \ .config("spark.sql.shuffle.partitions", 8) \ .config("spark.executor.memory", "8g") \ .config("spark.driver.memory", "6g") \ .getOrCreate() raw_path = "data/raw/yellow_tripdata_*.csv" df = spark.read.option("header", "true").option("inferSchema", "true").csv(raw_path)

这里用通配符一次性读取全部分片CSV。inferSchema会先扫描一遍数据体来推断字段类型，CSV里所有数值列都会变成int或double，时间字符串会保留为string，需要后面单独转换。如果省略inferSchema，所有列都会变成string，后续聚合时还得反复cast，反而多花时间。遇到字段格式不统一的文件，这一步多花几秒扫描是值得的。

接着做字段修正和分区键抽取，顺便把经纬度列重命名成统一的坐标列名：

df = df.withColumn("pickup_datetime", to_timestamp(col("tpep_pickup_datetime"))) \ .withColumn("dropoff_datetime", to_timestamp(col("tpep_dropoff_datetime"))) \ .withColumn("pickup_year", year(col("pickup_datetime"))) \ .withColumn("pickup_month", month(col("pickup_datetime"))) \ .withColumn("pickup_x", col("pickup_longitude")) \ .withColumn("pickup_y", col("pickup_latitude")) df.write \ .mode("overwrite") \ .partitionBy("pickup_year", "pickup_month") \ .format("parquet") \ .save("data/parquet/")

逻辑说明：tpep_pickup_datetime来自原始CSV，是字符串，通过to_timestamp统一成Spark的TimestampType；再用year、month抽取出行车发生的年月，作为分区键。写成Parquet时按年、月两级分区，后续Dask查某年某月的数据，只用读对应的分区目录，能跳过90%以上的无关文件。

注意：纽约出租车CSV里的时间字段默认是当地时间，没有任何时区标识。如果你的机器跑在UTC时区，后面用Dask统计“按小时分布”时会出现整体偏移，这个问题我在避坑章会展开说。

3.2 数据质量校验：ETL跑了不等于数据对了

ETL完成后，项目里有一段很朴素的校验逻辑，我看完之后觉得特别值得抄：

total_count = df.count() null_count = df.filter(col("pickup_datetime").isNull()).count() print(f"total={total_count}, null_pickup={null_count}") df.select( (col("fare_amount") < 0).alias("neg_fare"), (col("trip_distance") > 1000).alias("long_trip") ).groupBy("neg_fare", "long_trip").count().show()

这段代码不复杂，但它解决了一个真实痛点：大数据的ETL跑完，没人敢说原始CSV里没有脏数据。零元车费、负数金额、超过1000公里的行程，都是纽约出租车数据里真实存在过的脏数据。先算总数，再检查空值和异常值，确认数据形状符合预期，再继续往下做统计，这一步能省掉后面无数个“结果为什么不对”的深夜。

如果校验结果不对，项目里的做法是回到数据源，检查是不是某个月份的CSV格式和其他月不一致。比如有些CSV多一列，有些时间列格式不同，Spark会把这些文件读成null。这种情况在真正的生产数据里太常见了，我经常开玩笑说不校验就做统计等于盲人开车。

3.3 Dask统计模块：pandas语法直接跑200GB

等Parquet落好，02_stats_with_dask.ipynb就要上场了。这里的思路是用Dask DataFrame读Parquet，统计热力图之外的所有数值指标：

import dask.dataframe as dd ddf = dd.read_parquet( "data/parquet/", engine="pyarrow", columns=["fare_amount", "trip_distance", "pickup_year", "pickup_month"] ) monthly_avg = ddf.groupby(["pickup_year", "pickup_month"])["fare_amount"] \ .mean() \ .compute() print(monthly_avg)

逻辑说明：读Parquet时分目录扫描，只取计算需要的四列，用groupby做月均车费统计。Dask此时是惰性的，groupby只是构建计算图，真正跑起来是在调用compute()的瞬间。这个设计让你可以在notebook里先搭积木，确认逻辑没写错再触发计算，避免每改一次聚合就全量读一遍数据。

groupby后的compute有几个关键参数可以调。如果内存紧张，可以加scheduler="threads"用多线程执行；如果某个聚合输出特别大，还可以用split_out=4把结果拆成多个分区输出，减少单次内存压力。实际跑下来，计算每月平均车费，Dask只用了大约三分钟，比直接读原始CSV快出一个量级。

Parquet按列存储，读取时只加载你用到的那几列，这是Dask能读200GB数据的根本原因。

3.4 为什么ETL用Spark而不用Dask

这里有个值得讨论的选型问题：既然Dask也能读CSV，为什么ETL阶段用PySpark而不是直接用Dask？这个选择其实很有代表性。原始CSV有200GB，分散成多个月度文件，Dask虽然能读，但在解析带引号的文本字段、处理不规则行尾符时，容错能力不如Spark。Spark解析CSV时遇到坏行会默认忽略；Dask的read_csv遇到解析错误会直接抛异常，把整个任务中断。

另外，ETL阶段有大量的withColumn、to_timestamp这类逐行转换，Spark的Catalyst优化器会把这些转换串联成高效的执行计划。Dask的惰性图优化也有类似能力，但遇到多列Timestamp转换时，内存占用会明显高于Spark。所以项目里的经验是把一次性ETL交给Spark，之后所有可重复的查询交给Dask。我自己也照着这个模式写了几个数据集的分析，体感上比只用一个框架顺很多。

4. 可视化实战：Datashader绘制2亿+坐标点的时空热力图

4.1 为什么普通绘图库会在这一步崩掉

到了第三阶段，数据已经从200GB压缩到Parquet，单次查询的数据量可以控制在几GB内，但你要画的坐标点是两亿以上，Matplotlib和Plotly的scatter根本扛不住。它们会先把每个点都映射成屏幕上的一个图形对象，几亿个对象一次性创建，内存直接爆掉，或者渲染到交互界面时卡成幻灯片。

Datashader的做法完全不同：它先把坐标点按像素网格做聚合，每个像素格的最终颜色只取决于该格内落入多少个点，画出来的是栅格图而不是矢量散点图。坐标点数再多，最终网格是固定的，比如1200×800个像素，最多也就一百万格，内存占用是常数级的。一句话解释就是：Matplotlib画点是把每个点画出来，Datashader先统计每个像素里有多少点，再画统计结果。你要看的是分布密度，而不是单独某个点，Datashader正好为此设计。

4.2 聚合像素图：代码少于二十行

核心代码是这一段：

import datashader as ds import datashader.transfer_functions as tf import pandas as pd df = pd.read_parquet( "data/parquet/", engine="pyarrow", columns=["pickup_x", "pickup_y"] ) canvas = ds.Canvas( plot_width=1200, plot_height=800, x_range=(-74.5, -73.7), y_range=(40.5, 40.95) ) agg = canvas.points(df, "pickup_x", "pickup_y") img = tf.shade(agg, cmap=["#000000", "#ff9d00", "#ffffff"], how="eq_hist") img.save("heatmap.png")

逻辑说明：pickup_x和pickup_y是ETL阶段从原始经纬度重命名出的坐标列。Canvas定义的是屏幕上1200×800像素对应的地理范围，x_range和y_range是纽约市核心区的经纬度边界。canvas.points完成核心聚合：把DataFrame里亿级个点映射到网格上，统计每个格子内的点数量。tf.shade把聚合结果转成颜色图，how="eq_hist"是直方图均衡化。

如果不用eq_hist，绝大多数点集中的市中心会过曝，外围区域一片黑。eq_hist会把点密度做直方图拉伸，让密度低的区域也能显示出来。第一次画的时候我忘了加这个参数，整张图只有市中心是亮的，周围全黑，加上之后细节才出来。

关于Datashader参数，表里这三项最常改：

4.3 按月下钻：对比一年里每个月的出行热点

项目里还做了一个非常有用的下钻：把pickup_month按月过滤，生成12张月热度图，对比逐月出行热点变化。这段代码在make_viz.py里对应一个循环：

import datashader as ds import datashader.transfer_functions as tf import dask.dataframe as dd from dask.diagnostics import ProgressBar ddf = dd.read_parquet( "data/parquet/", engine="pyarrow", columns=["pickup_x", "pickup_y", "pickup_month"] ) def monthly_heatmap(month): ddf_part = ddf[ddf["pickup_month"] == month] canvas = ds.Canvas(1200, 800, x_range=(-74.5, -73.7), y_range=(40.5, 40.95)) agg = canvas.points(ddf_part, "pickup_x", "pickup_y") return tf.shade(agg, cmap=["black", "cyan", "white"], how="eq_hist") with ProgressBar(): imgs = [monthly_heatmap(m) for m in range(1, 13)]

这个循环里每个month对应一年的某个月。monthly_heatmap每次只查一个分区，Dask惰性计算保证月与月之间不会累积内存。ProgressBar是Dask的进度条上下文管理器，跑的时候能在notebook里看到已处理几百万行的实时进度，心里踏实很多。

下钻的价值在于：我能在图里直接看到夏夜酒吧区域高亮，午夜机场方向有长条形热点，这些是看统计数字感受不到的。Datashader的输出是栅格图，不能缩放查看单个点，但做全局分布观察已经足够。

实际运行monthly_heatmap时，单个月份的坐标点在千万级，canvas.points完成聚合大约需要几十秒。如果你想更快，可以在读取时用ddf.sample(frac=0.1)抽一折数据画预演图，确认样式没问题再全量跑。不要每次调试都跑全量，这是血的教训。

4.4 画图前要先把经纬度转成Web墨卡托

纽约出租车的经纬度是WGS84坐标，直接传给Datashader其实也能画，但和底图叠加时会偏移。常见做法是先用lnglat_to_meters转成Web Mercator投影，和在线地图底图对齐。这一步在项目脚本里体现为：

from datashader.utils import lnglat_to_meters df["pickup_x"], df["pickup_y"] = lnglat_to_meters( df["pickup_longitude"], df["pickup_latitude"] )

说明：WGS84的经度范围约在-74.5到-73.7，纬度在40.5到40.95，转成Web Mercator后坐标值会变成约-8293000到-8202000范围内的米制数。如果用原始经纬度画，结果不会出错但显示时坐标轴刻度怪怪的；如果未来要和在线地图叠加，转换是必须的一步。Datashader官方示例里经常直接传经纬度，但多一步转换能省掉很多后续麻烦。

5. 避坑记录：跑200 GB数据时最常见的五个翻车现场

下面是实打实跑出来的代价，每个坑都消耗过至少两个小时的调试时间。按“现象到原因到解决”的顺序写，每一条都能直接拿来对照。

5.1 CSV反复读取导致IO被打满

现象：ETL跑完后想改一个字段类型再跑一次，Spark重新读一遍200GB的CSV，磁盘IO马上100%，整台电脑卡到鼠标都移不动。

原因：CSV是文本格式，Spark每次加载都要完整解析200GB的字符串，而Parquet是列式二进制，读起来快一个数量级。我最初把ETL和统计放在同一个notebook里，每次启动都从头读CSV，完全没意识到重复读取的代价。

解决：把ETL抽成一次性脚本，输出data/parquet之后，所有后续查询只读Parquet。如果只是改字段类型，直接在Parquet上做覆盖写，不要回头碰原始CSV。我还给data/raw加了只读权限，逼自己不再直接读原始文件。

5.2 拍脑袋设置分区数导致OOM

现象：ETL在groupBy时报java.lang.OutOfMemoryError: Java heap space，异常堆栈指向Driver端，Notebook内核没过多久也跟着重启。

原因：我一开始把spark.sql.shuffle.partitions设成24，以为加大分区能并行更快，但单机只有4个核，24个并发task同时写内存，直接把driver内存顶爆。Spark理想情况是每个task吃一个核，分区数接近核数的2到3倍即可，不是越大越好。

解决：把分区数调回8，同时把executor内存提到8g。从那以后我每次设置分区数都会问自己一句：这个分区数在我的核数下，每个task分到多少数据？不是拍脑袋。

5.3 Dask惰性计算在什么时候真正触发计算

现象：某次查询，明明只是ddf[["fare_amount"]].describe()，却等了十分钟，一度怀疑Dask比Spark还慢。

原因：Dask的describe()和其他聚合一样是惰性的，调用的瞬间才开始触发完整计算，而describe本身要扫描整个Parquet的fare_amount列。我当时以为describe是简单操作应该秒回，实际它跟groupby一样走了全量扫描。

解决：先看ddf.npartitions和ddf.divisions，明确数据切成几块；小范围探索用.compute()前先取ddf.head(10000)验证逻辑，最后再跑全量。Dask不是慢，是我理解错了触发时机。

注意：Dask的describe()和groupby()一样惰性，别被“简单操作”骗了。

5.4 时区偏移让统计多出一小时

现象：按小时统计出行量时，凌晨0到1点数据异常低，半夜23到24点异常高，分布图多出一条奇怪的台阶。

原因：原始CSV里的tpep_pickup_datetime是当地时间，但我的机器在UTC时区。to_timestamp把字符串解析成无时区时间后，Spark按运行环境的时区解释，导致时间整体显示成UTC，和实际上车时间差了5小时。

解决：用spark.conf.set("spark.sql.sessionTimeZone", "America/New_York")，保证所有时间解析统一使用数据源当地时区。这个坑在跨时区分析时极具迷惑性，因为最终结果不是错，是整体偏移了一个常数，看图根本看不出来。

5.5 Jupyter Notebook内核重启后状态全丢

现象：ETL跑了一个半小时，去喝了杯咖啡回来发现notebook内核重启，所有变量全没了，数据要重跑。

原因：Spark任务内存吃紧时，会触发频繁GC，增加内核无响应并自动重启的概率。另外笔记本合上盖子休眠，也会中断Spark的本地连接。

解决：重要进度随时用df.write输出到Parquet，哪怕中间结果也先落盘，不依赖进程内变量。另外SparkContext所在进程不稳定，我把ETL从notebook迁到scripts/etl.py，用命令行方式执行，配合tmux运行，再也不怕内核重启了。

6. 二次开发：把这份分析改造成你自己的数据管道

这套项目最值得复用的部分是“CSV -> Parquet -> 聚合 -> 栅格图”的四段式，任何地理坐标数据分析都能直接套。想替换数据源，建议不要直接在notebook里改文件名，而是把data/raw的读取路径抽成命令行参数。比如在scripts/etl.py里：

parser.add_argument("--input", default="data/raw/*.csv") parser.add_argument("--output", default="data/parquet")

换成别的公开数据集时，只需改输入路径，字段清洗逻辑按新数据集的列名调整即可。

增量更新是另一个高频需求。出租车数据集每月更新一次，我的习惯是把ETL写成幂等任务：先判断目标分区是否已存在，存在就跳过，不存在才执行当月写入。Spark的partitionBy天然支持这一点，写之前先读一遍已有分区列表，做差集即可：

existing = spark.read.parquet("data/parquet") \ .select("pickup_year", "pickup_month").distinct().collect() existing_set = {(r.pickup_year, r.pickup_month) for r in existing} for year, month in [(2024, 1), (2024, 2)]: if (year, month) in existing_set: continue run_etl_for_month(year, month)

这个模式不限于出租车数据，任何按时间分区的日志类数据集都能用。

还有一件事是我养成的习惯：任何改动，先在Dask上抽样1%跑通，再全量跑Spark，全量结果保留一份分析口径的记忆笔记。从那以后我每次跑新数据都强制走一遍“先抽样、再全量、后落盘”的流程，省下来的调试时间远比测试数据消耗的时间多。希望帮到你。

本文还有配套的精品资源，点击获取