3、数据准备与上传：数据格式规范（CSV/Parquet/Image）、数据上传至云存储、数据版本控制（DVC/Delta Lake）

数据准备，说白了就是 ML 工作流的第一道关卡。我见过太多项目，模型选得再好，Pipeline 搭得再漂亮，最后全栽在数据上。嗯，这章我们就来聊聊数据格式、上传和版本控制——这三件事做好了，后面能省 80% 的麻烦。

3.1 数据格式规范：选对格式，事半功倍

你想想看，数据格式选错了会怎样？轻则多花几倍存储费，重则训练时 IO 直接卡死。我个人习惯，先问三个问题：数据多大？谁在用？要不要频繁读写？

3.1.1 CSV：最通用，但别滥用

CSV 的好处是简单，任何工具都能打开。但我在项目中遇到过，一个 10GB 的 CSV 文件，光是 pandas 读进来就花了 5 分钟，还占满了内存。

什么时候用 CSV？小数据集（< 1GB）、快速原型、或者跟非技术人员交换数据时。其他场景，我建议你慎重。

3.1.2 Parquet：列式存储，ML 的标配

Parquet 是我最常用的格式。它按列存储，压缩率高，还自带 schema。举个例子：

# 把 CSV 转成 Parquet
import pandas as pd

df = pd.read_csv('raw_data.csv')
df.to_parquet('clean_data.parquet', compression='snappy')

# 读取时只取两列，速度飞快
df_subset = pd.read_parquet('clean_data.parquet', columns=['feature_1', 'label'])

为什么快？因为 Parquet 只读取你需要的列，而不是整行。我在一个推荐系统项目里，把 200GB 的 CSV 转成 Parquet 后，存储降到 40GB，查询速度提升了 10 倍。

特性	CSV	Parquet
存储大小	大（无压缩）	小（列式压缩）
读取速度	慢（全量扫描）	快（列裁剪）
Schema 支持	无	有（强类型）
适用场景	小数据、临时分析	大数据、ML Pipeline

3.1.3 Image：别直接存原图

图像数据有个坑——你以为存了 JPG 就行，但训练时每次都要解码，IO 开销巨大。我曾经有个项目，数据是 50 万张图片，直接存 S3 上，训练时 CPU 全花在解码上，GPU 空转。

我的做法是：

• 小规模（< 10万张）：用 TFRecord 或 WebDataset 打包，顺序读取
• 大规模：用 Petastorm 或 NVIDIA DALI，直接在 GPU 上解码

3.2 数据上传至云存储：别用 cp，用工具

很多人上传数据，第一反应就是 aws s3 cp 或者 gsutil cp。嗯，小文件还行，大文件就等着哭吧。

3.2.1 并行上传与断点续传

我建议用 aws s3 sync 代替 cp，它支持增量同步和断点续传。举个例子：

# 同步本地 data/ 到 S3，只上传新增或修改的文件
aws s3 sync ./data/ s3://my-bucket/data/ --exclude "*.tmp" --include "*.parquet"

为什么用 sync？因为网络断了，重新跑一次就行，不会重复上传。我在一个跨国项目中，数据上传经常断，sync 救了我好几次。

3.2.2 数据分区：按时间或按类别

上传时就要想好分区策略。我习惯按日期分区：

s3://my-bucket/data/
├── year=2024/
│   ├── month=01/
│   │   ├── day=01/
│   │   │   ├── part-00001.parquet
│   │   │   └── part-00002.parquet
│   │   └── day=02/
│   └── month=02/
└── year=2025/

这样查询时只扫需要的分区，省时省钱。Spark 和 Presto 都原生支持这种分区格式。

3.3 数据版本控制：DVC 与 Delta Lake

代码有 Git 管理，数据呢？我见过最惨的案例：模型上线后效果变差，查了半天才发现是训练数据被人偷偷替换了。嗯，数据版本控制就是解决这个问题的。

3.3.1 DVC：轻量级的数据版本管理

DVC 的思路很简单——数据存在云存储，DVC 只记录元数据（哈希值、路径）。你可以在 Git 里追踪这些元数据，实现数据版本和代码版本的绑定。

# 初始化 DVC
dvc init

# 添加数据文件
dvc add data/train.parquet

# 配置远程存储
dvc remote add -d myremote s3://my-bucket/dvc-store

# 推送数据
dvc push

# 切换版本时，只需 git checkout 对应的 .dvc 文件，然后 dvc pull

我个人习惯，每次实验前先 dvc checkout，确保数据版本和代码一致。这样复现结果时，不会出现「在我机器上能跑」的尴尬。

3.3.2 Delta Lake：ACID 事务与时间旅行

Delta Lake 是 Databricks 推出的存储层，支持 ACID 事务、Schema 演化和时间旅行。说白了，你可以像操作数据库一样操作数据湖。

from delta import DeltaTable

# 读取 Delta 表
df = spark.read.format("delta").load("s3://my-bucket/delta-table/")

# 查看历史版本
delta_table = DeltaTable.forPath(spark, "s3://my-bucket/delta-table/")
history = delta_table.history().show()

# 时间旅行：回到 3 小时前的版本
df_old = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-15 10:00:00") \
    .load("s3://my-bucket/delta-table/")

为什么用 Delta Lake？因为你可以放心地让多个任务同时读写同一份数据，不会出现脏读。我在一个实时特征工程项目中，用 Delta Lake 管理每日更新的特征数据，再也没出现过数据不一致的问题。

特性	DVC	Delta Lake
定位	数据版本管理	数据湖存储层
ACID 事务	不支持	支持
时间旅行	通过 Git 间接实现	原生支持
适用场景	小团队、实验性项目	生产环境、多任务并发

3.4 本章知识体系

下面这张图，是我梳理的数据准备与上传的核心逻辑。你跟着这个流程走，基本不会出大问题。

数据准备这件事，看起来琐碎，但做扎实了，整个 ML 工作流就稳了一半。我个人经验是：花 30% 的时间在数据上，后面 70% 的时间会顺畅很多。别急着跑模型，先把数据管好。

---

公众号:蓝海资料掘金营，微信deep3321