企业官网建设流程全解析

简介：这份资源面向具备一定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；开了inferSchema，Spark 要额外扫一遍全量数据，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的第三个参数nullable=True表示允许空值，出租车数据里乘客数、费用确实可能缺失，设成 False 会在写入时报错。时间列直接用TimestampType，Spark 读的时候会按yyyy-MM-dd HH:mm:ss解析，如果你的原始格式不同，得在spark.read.csv里加timestampFormat参数。这一步做对，后面所有时间聚合都不用再 cast，省下的不只是代码行数，还有每次运行的解析开销。

3. 跑起来：从 CSV 到 Parquet 再到第一张热力图

3.1 环境准备与 SparkSession 参数

依赖不多，核心是 pyspark、dask、datashader、geopandas、matplotlib，Python 3.9 以上。装完之后第一件事不是跑 notebook，而是把 SparkSession 配好，项目里spark_session.py的写法值得抄：

from pyspark.sql import SparkSession def build_spark(app_name="nyc-taxi", driver_mem="8g", shuffle_partitions=200): 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 转分区 Parquet

01_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_width=800, plot_height=800, 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"], how="linear") img.to_pil().save("data/output/heatmap.png")

sample(0.01)是抽样，Datashader 本身能处理全量点，但toPandas()这一步会把数据拉进 driver 内存，百万行以上就有风险，抽样是稳妥做法。x_range和y_range必须和清洗时的坐标范围一致，否则图会偏。how="linear"是线性映射，如果热点区域对比不明显，可以换成how="eq_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")快，因为不用检查空值。这个查询的瓶颈在 shuffle：24 个分组键，但数据有几百个分区，每个分区都要把数据按 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向下取整，把连续坐标离散成格子编号，GRID=0.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", dpi=150)

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()报 OutOfMemoryError，driver 挂掉

原因：聚合结果比预期大。比如按网格 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 的数据真正跑出图来。

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