PySpark+Dask+Datashader：200GB出租车数据的大数据分析实战

简介一套围绕200 GB纽约出租车数据集展开的大数据可视化分析项目面向希望掌握分布式处理与大规模数据可视化的数据工程师、数据分析学习者。项目覆盖从原始数据存储、AWS EC2上部署Cloudera Hadoop集群、编写Hive SQL进行建表与记录统计到借助Python Datashader渲染高密度地理空间点的完整流程并涉及PySpark、Dask等分布式计算框架的衔接同时对比了有无Datashader时可视化海量出租车行程数据的效果差异可以帮助读者直观理解不同技术栈的取舍。压缩包共13个文件以Jupyter Notebook、Python脚本、Shell下载脚本和多个SQL查询文件为主辅以4张效果图与Markdown说明文档整体仅1.68MB轻量却结构完整便于按步骤复现实验环境。已有1383人学习下载适合需要参考真实规模数据项目拆解思路、搭建类似分析链路的中高级学习者。1. 200 GB NYC出租车数据不是下载问题是计算架构问题拿到一份 200 GB 的 NYC 出租车数据集第一反应往往是把 CSV 拖进 Pandas然后看着内存爆炸、内核重启最后怀疑人生。这个项目资源真正要解决的不是“怎么下载 200 GB”而是“拿到之后怎么算得动、画得出”。它围绕 PySpark、Dask、Datashader 三条技术线展开配合 Jupyter Notebook 环境把整套分析流程拆成了可复现的代码包。适合三类人刚接触大数据分析、想用真实数据练手的数据工程师被 Pandas 内存卡死的分析师以及想对比 Spark 和 Dask 在单机多核场景下性能差异的从业者。这份资源的价值不在于数据本身而在于它把踩坑过程也一并打包了。2. 选型为什么是 PySpark Dask Datashader而不是一台大内存机器2.1 先想清楚30 亿行数据能不能放进内存纽约出租车数据集按月份拆开是几十个 CSV合并之后大约 30 亿行、200 GB。很多人第一反应是“租个 512 GB 内存的云服务器硬扛”但算一笔账就明白了30 亿行每行算上 pickup 时间、落客时间、经纬度、费用字段在 Pandas 里一个 object 列就占 8 字节指针加上字符串本体实际占用经常膨胀到原来的 3 到 5 倍。按 4 倍估算200 GB 的 CSV 读进 Pandas 会变成 800 GB 左右512 GB 根本不够。Pandas 还有一个隐性限制它处理的是内存中的连续结构一旦数据超过内存就只能靠 swap而 swap 到硬盘之后性能断崖式下跌。所以选型的第一步不是选工具而是确定“数据能不能整体进内存”这个前提。这份资源的做法明确200 GB 数据不整体进内存而是交给 PySpark 做分布式聚合再用 Dask 做并行读取和预处理最后用 Datashader 做栅格化渲染绕开“一次性画上亿个点”的显卡和内存瓶颈。2.2 PySpark 与 Dask 的边界谁在算聚合谁在算可视化PySpark 和 Dask 都能做分布式计算但它们的定位完全不同。PySpark 的核心是 RDD 和 DataFrame它的调度器、Shuffle 机制、容错设计都是为“跨节点分布式计算”准备的。在这份资源里按区域、按时段的聚合统计比如算每小时的载客量、平均费用、平均行程时长交给 PySpark 最合适因为这类聚合需要 Shuffle而 Spark 的 Tungsten 优化和自适应查询引擎能自动处理数据倾斜和分区大小问题。Dask 则更轻量它运行在单个进程内通过多线程并行和惰性计算来模拟 DataFrame 操作。Dask 的强项是“并行 I/O 和内存映射”——可以直接读 Parquet、不做全量加载配合 Datashader 做流式渲染非常顺手。但 Dask 的 Shuffle 性能远不如 Spark如果强行拿 Dask 做大规模 GroupBy 聚合会在聚簇洗牌阶段卡很久。我的习惯是聚合比例大、需要稳定的容错 → PySpark数据探索、可视化前处理 → Dask两边各管一段不混着用。2.3 Datashader 的底层逻辑把“画图”变成“统计”Matplotlib 画 100 万个点已经很吃力而这份数据里仅一天的上客点就可能超过 50 万个直接画散点图会变成一坨黑色墨迹。Datashader 的思路完全不同它不画点而是把画布划分成网格统计每个网格单元里落了多少个点然后对网格做颜色映射。这一步把“矢量绘图”变成了“栅格化统计”本质上是用一个二维直方图替代散点图。这样做有两个直接好处第一计算量与点的数量无关只和画布分辨率有关所以 1 亿个点和 100 万个点的渲染时间是同一个量级第二输出结果是普通的 NumPy 数组可以直接转成图像不必依赖 WebGL 或 GPU。资源里 Datashader 的典型用法是先用 Dask 读入上客经纬度两列把无效坐标筛掉再传给 Datashader 的 Canvas 做聚合最后用 colorcet 的 color map 渲染出热力图。整个过程不产生中间放大文件也避开了“数据太大画不了”的窘境。2.4 三个工具的分工表工具负责阶段核心优势典型操作PySpark聚合统计Shuffle 稳定、容错好按区域/时段算载客量、费用均值Dask读取预处理并行 I/O、惰性计算、省内存读 Parquet、过滤非法坐标、列裁剪Datashader可视化栅格化聚合、与数据量解耦渲染热力图、等值线图Jupyter Notebook编排环境分步执行、快速验证逐段跑通分析流程这张表划清了边界PySpark 不负责画图Dask 不负责重聚合Datashader 不负责清洗。每个工具只做自己最擅长的一件事整条流水线跑起来才不互相拖累。3. 落地跑通从原始 Parquet 到一张可交互地图3.1 数据获取与格式转换先做一次“瘦身”纽约出租车数据的原始格式是 CSV按月份分文件。直接让 PySpark 读 CSV 也能跑但效率很低——CSV 需要逐行解析200 GB 读一遍可能耗时接近 20 分钟。资源里的第一步建议是先转成 Parquet并且只保留分析要用的列。from pyspark.sql import SparkSession spark SparkSession.builder.appName(nyc-taxi-convert) \ .config(spark.sql.adaptive.enabled, true) \ .config(spark.sql.adaptive.coalescePartitions.enabled, true) \ .getOrCreate() # 读取原始CSV只保留关键列避免后面反复全量扫描 df spark.read.csv( hdfs:///data/nyc_taxi/raw/*.csv, headerTrue, inferSchemaTrue ) selected_cols [ pickup_datetime, dropoff_datetime, pickup_longitude, pickup_latitude, trip_distance, fare_amount ] df.select(selected_cols).write.mode(overwrite) \ .parquet(hdfs:///data/nyc_taxi/parquet_monthly/)这段代码的核心动作是“列裁剪 格式转换”。inferSchemaTrue让 Spark 自动推断字段类型但注意CSV 里的 pick-up 时间如果格式不统一最好后面单独指定 schema避免推断时把时间戳解析成字符串。转成 Parquet 之后列式存储会让后续只读两列的查询大幅加速——只读 pickup 经纬度时Parquet 跳过其余列的数据块这在 200 GB 级别下是数量级的差异。提示如果你手头只有单机没有 HDFS把路径改成本地文件系统即可Spark 照样能读只是少了跨节点并行这一层优势。3.2 PySpark 做聚合统计按区域、按时段算载客量数据转成 Parquet 之后就可以做真正的聚合分析了。常见需求是算每个小时内不同区域的上客量用来观察早晚高峰的时空分布。这里的关键是“分桶粒度”——用 pickup_datetime 的小时做 GroupBy再按经纬度格网分区。from pyspark.sql.functions import hour, col, count, avg df spark.read.parquet(hdfs:///data/nyc_taxi/parquet_monthly/) # 新增“小时”列便于按时段聚合 df_hour df.withColumn(pickup_hour, hour(pickup_datetime)) # 按小时和粗略经纬度格网统计载客量、平均费用 result df_hour.groupBy(pickup_hour, col(pickup_longitude).cast(int), col(pickup_latitude).cast(int)) \ .agg(count(pickup_datetime).alias(trip_count), avg(fare_amount).alias(avg_fare)) \ .filter(col(trip_count) 100) \ .orderBy(pickup_hour, trip_count, ascending[True, False]) result.write.mode(overwrite) \ .parquet(hdfs:///data/nyc_taxi/agg_by_hour_grid/)逻辑说明这段先取pickup_datetime的小时部分生成新的维度列然后把经纬度直接 cast 成 int相当于把连续坐标离散成 1 度左右的网格这样聚合粒度可控也不至于把 30 亿行全部按唯一坐标拆开。filter(trip_count 100)把只有零星载客的偏远网格过滤掉避免结果里出现大量统计噪声。orderBy里用了两个排序字段这在 PySpark 中是允许的只要ascending列表长度匹配即可。参数说明经纬度直接cast(int)是一个很粗糙的分桶方式在纽约这种城市尺度下会有一定误差因为曼哈顿南北方向 1 度纬度约 111 公里实际热点会集中在少数几个格网里。更精确的做法是先把经纬度乘以 100 再 cast这样网格精度变成约 1.1 公里热点分布会清晰很多。资源里两种方式都有示例建议至少用round(lon * 100)这个粒度。3.3 Dask Datashader 渲染热力图不上点直接上画布聚合统计得到的是表格但可视化需求往往是“把某一周的上客点画成热力图”。这一步用 PySpark 再算一次坐标明细会很笨拙——毕竟单机环境下拉全量坐标到内存不现实。正确姿势是 Dask 惰性读取 Datashader 栅格化。import dask.dataframe as dd import datashader as ds import datashader.transfer_functions as tf from colorcet import fire # 用Dask读取前一个环节生成的Parquet只取经纬度两列 ddf dd.read_parquet( hdfs:///data/nyc_taxi/parquet_monthly/, columns[pickup_longitude, pickup_latitude, pickup_datetime] ) # 筛选出有效坐标过滤掉默认的(0, 0)和越界值 x_range (-74.05, -73.75) y_range (40.55, 40.95) ddf ddf[(ddf[pickup_longitude] x_range[0]) (ddf[pickup_longitude] x_range[1]) (ddf[pickup_latitude] y_range[0]) (ddf[pickup_latitude] y_range[1])] # 建画布并一次性栅格化 canvas ds.Canvas(plot_width800, plot_height600, x_rangex_range, y_rangey_range) agg canvas.points(ddf, pickup_longitude, pickup_latitude) img tf.shade(agg, cmapfire, howeq_hist)逻辑说明dd.read_parquet是惰性的只是建立了计算图真正触发计算的是最后一行tf.shade。这样就避免了“先把数据全部读进内存再处理”的套路。eq_hist是 Datashader 的直方图均衡化着色方式它能自动拉伸颜色区间让低密度区域也可见而不是被高密度中心区压制。fire色带从黑到黄再到白适合表达密度递增。参数说明plot_width800, plot_height600直接决定了栅格化的网格数——每个像素一个格子计算量与此成正比。我一般不建议超过 1200 宽除非你有很强的 CPU否则收益递减。如果发现颜色分布太集中改howeq_hist而不是去调 cmap如果觉得密度溢出严重可以改howlog它会对统计值取对数再映射颜色。3.4 Jupyter Notebook 里的集成共享 SparkContext 与内存释放在 Notebook 里把 PySpark 和 Datashader 串起来时最容易出问题的环节是资源管理。PySpark 默认申请完 executor 内存后不会自动释放而后面的 Dask 线程池又会再占一份内存两者叠加容易导致单机 OOM。# 在Notebook最前面启动Spark并明确限制内存占用 spark SparkSession.builder \ .master(local[4]) \ .appName(nyc-taxi-notebook) \ .config(spark.driver.memory, 8g) \ .config(spark.executor.memory, 4g) \ .config(spark.local.dir, /tmp/spark-tmp) \ .getOrCreate() # 用一个context manager包裹Dask客户端确保用后即焚 from dask.distributed import Client, LocalCluster cluster LocalCluster(n_workers4, threads_per_worker2, memory_limit8GB, processesFalse) client Client(cluster) # 中间结果用完立刻清理 client.restart() client.close() cluster.close()逻辑说明master(local[4])表示在单机开 4 个 CPU 核做任务并行适合资源有限的实验环境spark.local.dir是关键——Spark 在 Shuffle 过程中会写临时文件默认写在系统临时目录如果磁盘空间不足会直接报错。Dask 的LocalCluster里processesFalse表示用线程模式这样和 Jupyter 进程共享内存空间内存管理更可控。参数说明memory_limit8GB是给 Dask 每个 worker 设的内存上限如果数据分片超过这个值Dask 会走磁盘 spill速度变慢但至少不崩。spark.executor.memory4g和 Dask 的 8GB 加在一起差不多 12GB如果你的机器只有 16GB 内存建议把 Spark 的 executor 降到 2g或者等 Spark 任务跑完再启动 Dask二选一来不要同时跑重任务。4. 200 GB 避坑指南四个让我翻车的真实教训4.1 现象作业卡死不动Spark UI 显示大量任务 Pending第一次跑聚合时我盯着 Spark UI 看了十分钟发现所有任务都停在 Pending 状态没有任何执行进度。排查发现原因是输入文件是几千个小 CSVSpark 默认按文件分片导致每个分片任务只处理几 MB 数据调度器被海量小任务淹没资源全耗在任务调度上而不是计算上。解决先把小 CSV 合并成少量大文件或者直接统一转成 Parquet 并设置合理的分区数。我用的方法是repartition(200)把数据重分成 200 个左右的分区让每个分区大约 1 GB任务数量立刻降了两个量级。从那以后我只要遇到 Pending 成片第一反应就是看输入文件大小分布而不是怀疑集群配置。4.2 现象内存不爆但聚合跑了 40 分钟还没结束spill 到磁盘有次我也是做按小时的 GroupBy数据只有 60 GB但跑一个小时的聚合就写了 200 GB 的 shuffle 临时文件任务慢得像在爬。原因有两个一是按小时聚合的 Key 基数太大Shuffle 阶段要把同一个小时的所有记录拉到同一节点二是没有做预聚合所有明细记录都参与了网络传输。解决PySpark 里开了spark.sql.adaptive.enabledtrue让优化器自动合并小分区同时我对小时和经纬度网格做了“两级聚合”预处理——先按经纬度网格粗聚合一次再在粗聚合结果上按小时二次聚合中间数据量直接缩小到原先的 1/20。这种“先粗后细”的聚合策略在数据量上了 100 GB 之后几乎是必选项。4.3 现象Datashader 渲染出来整张图是黑的只有少数几个亮点我对着渲染结果看了半天黑色的底图上只有零星几个黄点直觉告诉我数据分布出了问题。排查后发现在 Dask 筛选pickup_longitude时我用了-74.0这个边界但数据里有大量坐标为0的无效记录它们被保留下来后出现在画布之外更严重的是部分记录的经纬度被存成了字符串类型比较时发生了隐式类型转换把有效坐标全过滤掉了。解决改造了过滤条件——先显式把经纬度列.astype(float64)再筛x_range和y_range同时补一条(pickup_longitude ! 0) (pickup_latitude ! 0)的条件彻底清掉无效坐标。之后渲染出的热力图密度分布正常了。这个坑极其隐蔽因为程序不报错只是结果看起来莫名其妙。4.4 现象Jupyter 内核被 killed进程直接消失这是最有挫败感的错误。跑 PySpark 聚合一切正常但一旦运行 Datashader 的tf.shadeNotebook 内核瞬间消失前面的计算结果全部作废。原因很直接Datashader 触发了 Dask 的完整计算而这时 Spark 的 executor 还占着大半内存Dask 想申请新内存时操作系统直接 OOM把整个 Jupyter 进程杀了。解决最终是用“错峰执行”兜底——在 Notebook 里先用spark.stop()手动关掉 Spark再启动 Dask 跑渲染。这看起来笨但确实最稳。如果你非要同时跑就把 Spark 的 executor 内存压到 2gDask 的 worker memory 降到 4g牺牲并行度换进程存活。从那以后我只要在 Notebook 里混用两个引擎就强制先用spark.stop()清场。5. 验证与进阶把聚合结果接回 Datashader跑通闭环5.1 先验证基线百万级随机点的性能基准拿到资源后别急着全套复现我建议先跑一个最基础的 Datashader 基准确认环境没问题再上真数据。方法是用 NumPy 随机生成 100 万个经纬度点走一遍canvas.points tf.shade记录耗时。import numpy as np import datashader as ds import datashader.transfer_functions as tf from colorcet import coolwarm import time n_points 1_000_000 lon np.random.uniform(-74.05, -73.75, n_points) lat np.random.uniform(40.55, 40.95, n_points) canvas ds.Canvas(plot_width800, plot_height600) t0 time.time() agg canvas.points(canvas.source(lon, lat), lon, lat) img tf.shade(agg, cmapcoolwarm) print(frendered {n_points} points in {time.time() - t0:.2f}s)这段代码拆开看有两个关键点canvas.source(lon, lat)把 NumPy 数组直接包装成 Datashader 可识别的数据源省掉了 DataFrame 的开销整个渲染过程只对 800×600 网格做统计所以耗时应稳定在 2 秒以内。如果跑出来的耗时超过 5 秒说明你的单核 CPU 性能偏弱或者 Datashader 版本有编译问题需要先排查环境再继续。验证通过之后再把第 3 章里的真数据流程走一遍正常应该比基准慢 3 到 5 倍属于合理范围。如果慢得离谱回头看是不是 Dask 分片太多、读 Parquet 的列没裁剪干净。5.2 进阶把 PySpark 聚合结果当成 Datashader 的输入层一个常被忽略的组合场景是先用 PySpark 算出网格化的聚合结果比如每个 100m 格网的载客量然后把这个聚合结果倒回 Datashader 做等值线或地形图渲染。这样既绕开了原始明细数据又把 Spark 的算力和 Datashader 的渲染优势串在一起。from pyspark.sql.functions import round # 在Spark里把经纬度缩放到100米网格 df df.withColumn(grid_lon, round(col(pickup_longitude) * 1000, 0)) df df.withColumn(grid_lat, round(col(pickup_latitude) * 1000, 0)) grid_count df.groupBy(grid_lon, grid_lat) \ .count().filter(col(count) 50) \ .toPandas() # 聚合结果通常在百万行以内可以直接转Pandas再放进Datashader import datashader as ds import datashader.transfer_functions as tf canvas ds.Canvas(plot_width900, plot_height650) agg canvas.points(grid_count, grid_lon, grid_lat, aggds.count(count)) img tf.shade(agg, cmapfire, howeq_hist)逻辑说明第一步把经纬度分别乘以 1000 取整数相当于把连续坐标离散到约 100 米的网格GroupBy 之后的数据量会从 30 亿行降到几百万行toPandas()转回本地内存完全可行。第二步 Datashader 的aggds.count(count)表示对 Spark 算好的 count 字段做加权聚合而不是对点做普通计数这样渲染出来的热力图密度反映的是真实载客强度。这个组合是我后来用这个资源最顺手的方式重聚合交给 Spark渲染交给 Datashader两者之间只传递一个小巧的网格表。从那以后我每次拿到超大数据集都强制自己先走一遍“先分区小样 → 跑通 Pipeline → 再上全量”的三步流程——小样数据能跑通全量数据就跑不崩反过来小样跑不通直接冲全量只会收获一个内核退出的教训。希望帮到你。本文还有配套的精品资源点击获取