简介这份资源面向具备一定Python与大数据基础、希望上手真实大规模数据处理的学习者围绕约200GB的NYC出租车数据集演示如何借助AWS EC2搭建Hadoop集群完成存储与处理并用Datashader对海量数据做可视化呈现。包内共13个文件以Jupyter Notebook为主配合Hive建表与统计查询脚本、数据下载脚本及若干可视化结果图另有说明文档压缩包约1.68MB便于按流程复现实验。内容涵盖集群部署、数据导入、无Datashader与使用Datashader的对比可视化等环节可帮助读者理解大数据场景下的绘图瓶颈与优化思路掌握PySpark、Dask等工具的协同使用。目前已有1383人学习下载适合作为大数据可视化项目的实践参考。1. 从一块 200 GB 的硬盘说起这个项目到底在算什么去年帮一个做城市交通研究的朋友收拾数据他丢过来一块移动硬盘里面躺着 200 GB 的 NYC 出租车行程记录CSV 一行一行堆着用 Excel 打开直接卡死用 pandas 读前 100 万行内存就飙到 8 GB。他想要的东西其实很朴素一天里哪个小时打车最难、机场线路的时长分布、周末和工作日的差异、热点区域在地图上的密度。这些需求用单机 pandas 做不了用 Spark 集群又太重。这个项目就是冲着这个夹缝来的——用 PySpark 或 Dask 做分布式聚合用 Datashader 把上亿个 GPS 点渲染成一张能看的热力图全程跑在 Jupyter Notebook 里一台 16 GB 内存的笔记本就能啃下 200 GB 的原始数据。适合谁适合手里有大数据集、但不想搭集群、又想保留交互式探索体验的开发者。下面我按「它怎么组织、怎么跑起来、怎么改」拆一遍。2. 技术栈与目录结构为什么是 PySpark Dask Datashader 这套组合2.1 三个引擎各管一段别指望一个工具通吃这个项目的技术选型不是拍脑袋而是按数据规模分层的。原始数据 200 GB单机 pandas 的瓶颈不在算力而在内存——它要求整个 DataFrame 驻留内存200 GB 压缩后解压可能到 400 GB 以上任何 16 GB 机器都扛不住。所以第一层必须用能流式处理、能 spill 到磁盘的引擎PySpark 的 DataFrame API 和 Dask 的 partition 机制都满足。两者的分工在这个项目里很清晰PySpark 负责重活也就是读 Parquet、做 groupby 聚合、算分位数这类需要 shuffle 的操作它背后有 Catalyst 优化器和 Tungsten 执行引擎SQL 风格的聚合写得越规整越快Dask 负责轻活和衔接比如把聚合后的小结果通常几十 MB转成 pandas DataFrame或者对中等规模数据做并行 map。Datashader 则是最后一步的渲染器它不画散点图而是把坐标点先栅格化成像素网格再上色所以哪怕十亿个点也不会让浏览器卡死——这是它和 matplotlib 散点图的本质区别。我一般会这样判断用哪个数据量超过内存 3 倍以上、且操作是聚合类的走 PySpark数据量在内存 1 到 3 倍之间、需要反复迭代调试的走 Dask已经聚合到百万行以内、要出图的直接 pandas Datashader。项目里的 notebook 基本按这个边界切分读者改的时候别把边界搞混否则会出现「Spark 里做小数据可视化」这种又慢又难调的翻车现场。2.2 目录结构notebook 按分析阶段编号数据分层存放项目目录大致长这样我按实际拆包的经验还原nyc-taxi-analysis/ ├── notebooks/ │ ├── 01_data_ingest.ipynb # 原始 CSV 转 Parquet建分区 │ ├── 02_cleaning.ipynb # 清洗时间范围、坐标越界、异常时长 │ ├── 03_aggregation.ipynb # PySpark 聚合小时/区域/线路统计 │ ├── 04_visualization.ipynb # Datashader 热力图与分布图 │ └── 05_export.ipynb # 结果导出 CSV/PNG ├── src/ │ ├── spark_session.py # SparkSession 构建与参数 │ ├── schema.py # 显式 schema 定义 │ └── geo_utils.py # 坐标过滤与网格映射 ├── data/ │ ├── raw/ # 原始 CSV不入库 │ ├── parquet/ # 分区后的 Parquet │ └── output/ # 聚合结果 ├── conf/ │ └── spark-defaults.conf # 内存与并行度配置 └── requirements.txt这个结构的关键在于data/parquet/这一层。原始 CSV 只读一次转成按year/month分区的 Parquet 之后后续所有 notebook 都读 Parquet列式存储加分区裁剪能把扫描量砍掉一个数量级。src/里放的是被多个 notebook 复用的逻辑尤其是 schema 和 SparkSession 构建——这两块最容易在 notebook 里复制粘贴出不一致抽出来是血泪经验。2.3 显式 schema别让 Spark 猜类型读 CSV 时如果不开inferSchema所有列都是 string后面做时间运算全得 cast开了inferSchemaSpark 要额外扫一遍全量数据200 GB 就是双倍 IO。正确做法是手写 schema项目里schema.py干的就是这事from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType # 显式定义 schema避免 inferSchema 触发全量扫描 taxi_schema StructType([ StructField(vendor_id, StringType(), True), StructField(pickup_datetime, TimestampType(), True), StructField(dropoff_datetime, TimestampType(), True), StructField(passenger_count, IntegerType(), True), StructField(trip_distance, DoubleType(), True), StructField(pickup_longitude, DoubleType(), True), StructField(pickup_latitude, DoubleType(), True), StructField(dropoff_longitude, DoubleType(), True), StructField(dropoff_latitude, DoubleType(), True), StructField(fare_amount, DoubleType(), True), ])StructField的第三个参数nullableTrue表示允许空值出租车数据里乘客数、费用确实可能缺失设成 False 会在写入时报错。时间列直接用TimestampTypeSpark 读的时候会按yyyy-MM-dd HH:mm:ss解析如果你的原始格式不同得在spark.read.csv里加timestampFormat参数。这一步做对后面所有时间聚合都不用再 cast省下的不只是代码行数还有每次运行的解析开销。3. 跑起来从 CSV 到 Parquet 再到第一张热力图3.1 环境准备与 SparkSession 参数依赖不多核心是 pyspark、dask、datashader、geopandas、matplotlibPython 3.9 以上。装完之后第一件事不是跑 notebook而是把 SparkSession 配好项目里spark_session.py的写法值得抄from pyspark.sql import SparkSession def build_spark(app_namenyc-taxi, driver_mem8g, shuffle_partitions200): return (SparkSession.builder .appName(app_name) .config(spark.driver.memory, driver_mem) .config(spark.sql.shuffle.partitions, shuffle_partitions) .config(spark.sql.parquet.compression.codec, snappy) .config(spark.local.dir, /tmp/spark-tmp) # 溢出到磁盘的目录 .getOrCreate())driver_mem设成 8g 是给本地模式留的余量如果你机器只有 16 GB别超过 10g否则系统会开始 swap速度反而断崖下跌。shuffle_partitions默认是 200对小数据偏多、对大数据偏少200 GB 数据我一般按「每分区 128 MB」估算大概 300 到 500项目里给 200 是个保守起点跑聚合时如果发现 task 倾斜再往上调。spark.local.dir指向一个剩余空间大的盘shuffle 溢出全落这里指向根分区很容易把系统盘写满——这个坑我踩过半夜跑任务把/写爆机器直接失联。3.2 原始 CSV 转分区 Parquet01_data_ingest.ipynb的核心就一段但参数很讲究from src.schema import taxi_schema from src.spark_session import build_spark spark build_spark() df (spark.read .option(header, true) .schema(taxi_schema) .csv(data/raw/*.csv)) # 按年月分区写入后续查询可做分区裁剪 (df.withColumn(year, year(pickup_datetime)) .withColumn(month, month(pickup_datetime)) .repartition(year, month) .write .mode(overwrite) .partitionBy(year, month) .parquet(data/parquet/))repartition(year, month)在写之前按分区键重分布保证每个输出目录下的文件数合理避免出现几千个小文件——小文件是 Spark 读性能的头号杀手每个文件都要开一次 IO。mode(overwrite)方便重跑但注意它会清空目标目录别指向有别的数据的路径。写入格式选 Parquet 加 snappy 压缩压缩比大概 3 到 5 倍200 GB CSV 转完通常落在 40 到 60 GB这个体积对单机就友好多了。3.3 清洗三类脏数据必须先处理02_cleaning.ipynb处理的是出租车数据的老三样时间越界、坐标越界、时长异常。时间上原始数据里混着 2001 年、2090 年这种明显错误的记录坐标上经纬度落在纽约市范围之外的大致经度 -74.3 到 -73.7纬度 40.5 到 41.0要剔除时长上小于 1 分钟或大于 5 小时的行程基本是计价器故障或司机忘关表。from pyspark.sql.functions import col, unix_timestamp clean (df .filter((col(pickup_datetime) 2015-01-01) (col(pickup_datetime) 2020-01-01)) .filter((col(pickup_longitude).between(-74.3, -73.7)) (col(pickup_latitude).between(40.5, 41.0))) .withColumn(duration_min, (unix_timestamp(dropoff_datetime) - unix_timestamp(pickup_datetime)) / 60) .filter((col(duration_min) 1) (col(duration_min) 300)))between是闭区间边界值会被保留实际业务里边界附近的点也常是噪声可以再收紧一点。unix_timestamp返回秒除以 60 得分钟注意这里没处理跨时区——纽约数据统一用当地时间不用管。清洗完建议先count()一下看看剩多少行如果掉得超过 20%说明过滤条件可能写错了别急着往下跑。3.4 用 Datashader 出第一张密度图聚合完之后把 pickup 坐标抽出来画热力图。Datashader 的用法和 matplotlib 差别很大它分两步先Canvas定义画布和聚合方式再shade上色。import datashader as ds import datashader.transfer_functions as tf import pandas as pd # 聚合结果已降到百万行以内转 pandas pts spark.sql( SELECT pickup_longitude AS x, pickup_latitude AS y FROM parquet.data/parquet/ WHERE pickup_longitude IS NOT NULL ).sample(0.01).toPandas() # 采样 1% 控制内存 canvas ds.Canvas(plot_width800, plot_height800, x_range(-74.3, -73.7), y_range(40.5, 41.0)) agg canvas.points(pts, x, y, ds.count()) img tf.shade(agg, cmap[white, yellow, red], howlinear) img.to_pil().save(data/output/heatmap.png)sample(0.01)是抽样Datashader 本身能处理全量点但toPandas()这一步会把数据拉进 driver 内存百万行以上就有风险抽样是稳妥做法。x_range和y_range必须和清洗时的坐标范围一致否则图会偏。howlinear是线性映射如果热点区域对比不明显可以换成howeq_hist做直方图均衡视觉效果会强很多。这张图出来基本就能看出曼哈顿中城和两个机场的密度峰值项目的第一阶段目标就算达成了。4. 核心模块拆解聚合逻辑与可视化链路的关键实现4.1 小时级聚合一个 groupby 背后的 shuffle 代价「一天里哪个小时最难打车」这个需求落到代码就是按 pickup 小时分组计数。看起来简单但 200 GB 数据做 groupby 会触发全量 shuffle这是整个项目最重的一步from pyspark.sql.functions import hour, count, avg hourly (spark.read.parquet(data/parquet/) .withColumn(pickup_hour, hour(pickup_datetime)) .groupBy(pickup_hour) .agg(count(*).alias(trip_count), avg(duration_min).alias(avg_duration)) .orderBy(pickup_hour))hour()提取 0 到 23 的整数count(*)比count(col)快因为不用检查空值。这个查询的瓶颈在 shuffle24 个分组键但数据有几百个分区每个分区都要把数据按 key 重新分发。如果spark.sql.shuffle.partitions设得太大比如 2000会产生大量小任务调度开销超过计算本身设得太小比如 50单个 task 处理的数据量过大容易 OOM。我的经验是看 Spark UI 里 shuffle read 的每个 task 大小控制在 100 到 200 MB 之间最舒服。聚合结果只有 24 行但它是从 200 GB 里算出来的。这里有个优化点如果只需要计数可以用spark.sql.shuffle.partitions调小到 100 左右因为分组键少不需要那么多分区如果还要算分位数比如 P95 时长那就得用percentile_approx它比精确分位数省资源误差可控。4.2 区域网格映射把经纬度变成可聚合的格子热力图之外项目还想按区域统计比如「哪个网格的上车点最多」。直接对经纬度做 groupby 会得到几十万个分组不现实所以要先把坐标映射到固定网格from pyspark.sql.functions import floor GRID 0.01 # 约 1 公里网格 gridded (df .withColumn(grid_x, floor(col(pickup_longitude) / GRID)) .withColumn(grid_y, floor(col(pickup_latitude) / GRID)) .groupBy(grid_x, grid_y) .agg(count(*).alias(cnt)))floor向下取整把连续坐标离散成格子编号GRID0.01在纽约纬度上大约对应 1.1 公里这个粒度对城市分析够用。格子太细0.001会导致分组数爆炸太粗0.1又看不出街区差异。映射完之后 groupby 的分组数从几十万降到几千shuffle 压力小一个量级。这个技巧在geo_utils.py里封装成了函数改网格大小只改一个常量。4.3 Datashader 与 matplotlib 的分工什么时候不该用它项目里有个容易误用的点Datashader 适合海量点的密度渲染但不适合画折线图、柱状图。04_visualization.ipynb里小时分布用 matplotlib 画柱状图只有坐标密度图才用 Datashader。原因是 Datashader 的输出是栅格图像没有坐标轴语义你要在图上标注「凌晨 3 点」这种信息还得回到 matplotlib 叠加。import matplotlib.pyplot as plt pdf hourly.toPandas() # 24 行随便转 plt.bar(pdf[pickup_hour], pdf[trip_count]) plt.xlabel(hour of day) plt.ylabel(trip count) plt.savefig(data/output/hourly.png, dpi150)toPandas()在这里完全安全因为只有 24 行。判断标准很简单聚合后行数在十万以内pandas matplotlib 足够超过百万且是散点/密度类才上 Datashader。把这两个工具用反了要么图糊成一团要么渲染慢到无法交互。4.4 结果导出Parquet 与 PNG 分开存05_export.ipynb把聚合结果写成 Parquet 供后续复用图存 PNG。这里有个细节聚合结果虽然小但也建议写 Parquet 而不是 CSV因为下游如果再用 Spark 读Parquet 有 schema 且列式读取快CSV 每次读都要重新推断类型容易出岔子。图存 PNG 时dpi设 150 到 200再高文件体积涨得快屏幕上看不出差别。5. 避坑与排查五个真实踩过的坑5.1 现象任务跑到 90% 卡住不动Spark UI 显示某个 task 特别慢原因数据倾斜。出租车数据里曼哈顿的行程量远超其他区按区域 groupby 时某个 key 对应的数据量可能是其他的几十倍单个 task 处理不完。解决先df.groupBy(grid_x).count().orderBy(desc(count)).show(10)确认倾斜键然后对热点键加随机前缀打散再聚合或者直接调大spark.sql.shuffle.partitions让每个 task 分到的数据少一点。我一般先试调分区数不行再上加盐。5.2 现象toPandas()报 OutOfMemoryErrordriver 挂掉原因聚合结果比预期大。比如按网格 groupby 后有几百万行全拉到 driver 内存就爆了。解决在toPandas()之前先limit()或sample()或者把结果写成 Parquet 再用 Dask 分块读。别迷信「聚合后数据一定小」分组键选细了照样大。5.3 现象Parquet 目录下生成几千个小文件后续读取极慢原因写入前没做repartition或者分区键基数太大比如按小时分区一年就是 8760 个目录。解决写入前按目标分区数repartition分区键选基数适中的列年月就够单个文件控制在 128 MB 左右。已经生成小文件的可以读进来repartition再重写一遍。5.4 现象Datashader 出的图一片空白或全黑原因x_range/y_range和数据实际范围不匹配或者坐标列有 NaN 没过滤。解决先pts[[x,y]].describe()看实际范围确认Canvas的范围覆盖数据NaN 用dropna()去掉。另外tf.shade的cmap如果只给一个颜色输出就是单色至少给两个。5.5 现象notebook 重启后 SparkSession 报「already stopped」原因getOrCreate()在同一个 kernel 里多次调用前一个 session 被stop()过。解决把build_spark()的返回值存成全局变量复用别在每个 cell 里重新建确实要重建时先spark.stop()再建。这个坑在反复调试时特别常见养成「一个 notebook 一个 session」的习惯。6. 二次开发换数据集、加指标、接调度拿到这个项目想改成自己的最省力的路径是换数据不动骨架。比如换成共享单车的行程数据字段名不同但结构类似改schema.py的列名映射、改清洗里的坐标范围、改聚合里的小时提取逻辑三个文件就能跑通。我一般会先写一个config.py把路径、坐标范围、网格大小、时间窗口全抽成常量这样换数据集只改配置不改代码验证起来也快——跑一遍01到03看行数掉幅是否在合理区间再看聚合结果的分组数是否正常两步就能判断适配对不对。加指标的话项目现有的聚合是计数和均值想加「费用中位数」「平均速度」这类在agg里追加percentile_approx(fare_amount, 0.5)和avg(col(trip_distance) / (col(duration_min) / 60))就行。注意速度计算要过滤duration_min为 0 的记录否则除零会得到 inf写进 Parquet 后下游读出来是脏数据。加完指标建议单独跑一个小样本比如只读一个月的分区验证数值合理性别直接上全量全量跑一次半小时错了重来很浪费时间。想接调度的话把 notebook 里的逻辑抽成src/jobs/下的 Python 脚本用spark-submit提交参数通过argparse传。notebook 适合探索生产跑批还是脚本稳。抽的时候注意把display()、show()这类交互调用去掉换成日志输出否则在无头环境里会报错。有个习惯我保持了几年每次改完聚合逻辑先在一个月的分区上跑一遍把结果和改之前对比差异超过预期就停下来查。这个「小样本回归」救过我很多次因为大数据集上的错误往往不是报错而是数值悄悄偏了等全量跑完才发现就晚了。希望这个项目拆解能帮你少走几步弯路把 200 GB 的数据真正跑出图来。本文还有配套的精品资源点击获取
