2.1 Dataset：把数据变成可调度的流水线

2.1 Dataset：把数据变成可调度的流水线

本节摘要：tf.data.Dataset 是 TensorFlow 数据供给的统一抽象：它把"数据从哪来、怎么加工、怎么成批"声明成一条惰性求值的元素流，直到被迭代那一刻才真正执行。本节用加州房价数据跑通一条最短管道，讲透"声明与执行分离"的调度本质，并给出 shuffle、batch、repeat 三个基本变换的语义与先后顺序的门道。管道思维是本章所有后续内容的骨架。

学习目标

阅读完本节，你应当能够：

1. 解释 Dataset 的惰性求值：变换声明不执行，迭代才开机；
2. 用 from_tensor_slices 加四个变换搭出标准训练管道；
3. 说出 shuffle、batch、repeat 的顺序差异对训练的实际影响；
4. 用 take 与 as_numpy_iterator 观测管道产出而不触发全量执行。

为什么训练循环不该自己写

不用 tf.data，训练喂料长这样：一个 for 循环，每步从数组切一片、转张量、可能再做标准化。问题是三重的：切数据在 Python 层做，GIL 锁住了解析与计算的重叠机会；打乱、成批、循环轮次的逻辑散落在循环代码里，换个数据集就要重写一遍；GPU 算完一步等着下一步数据时，没人提前备料——计算与读取串行，设备空转。tf.data 的解法是把整条上料流程声明成管道，交给框架的 C++ 运行时调度：读取、加工、打乱、成批各占一个工位，可以并行、可以重叠、可以被预取。

先跑通最短管道，用加州房价数据：

import tensorflow as tf
import numpy as np
from sklearn.datasets import fetch_california_housing

housing = fetch_california_housing()
X = housing.data.astype("float32")     # 20640 条样本，8 个特征
y = housing.target.astype("float32")

# 最短管道：切片 → 打乱 → 成批 → 预取
ds = tf.data.Dataset.from_tensor_slices((X, y))
ds = ds.shuffle(buffer_size=1000).batch(64).prefetch(tf.data.AUTOTUNE)

for x_batch, y_batch in ds.take(2):     # take(2)：只迭代前两个批次
    print(x_batch.shape, y_batch.shape)
# 输出：
# (64, 8) (64,)
# (64, 8) (64,)
​

五行声明，产出的就是 fit 直接可吃的批次流。关键是理解这五行里没有一行在声明时执行——from_tensor_slices 只是把数组的视图登记进计划，shuffle、batch 各往计划尾部追加一环，迭代器被创建时管道才开机。

声明与执行分离：管道的执行真相

Dataset 对象本质上是一张声明式的取货计划。验证这一点最直观的办法是打印它：

raw = tf.data.Dataset.from_tensor_slices((X, y))
pipe = raw.shuffle(1000).batch(64)
print(pipe)
# 输出（示意）：
# <ShuffleDataset buffer_size: 1000, seed: None, reshuffle_each_iteration: true,
#   ... <BatchDataset batch_size: 64, drop_remainder: false,
#     ... <TensorSliceDataset shapes: ((8,), ()), types: (tf.float32, tf.float32)>>>
# 类型名由内向外嵌套——计划的结构一目了然

# 同一份数据可以派生多条独立管道
train_pipe = raw.shuffle(1000).batch(64)
val_pipe = raw.batch(256)               # 验证集不打乱、大批次
​

执行时机有一条铁律：只有在创建迭代器或传入 fit 之后，变换才逐环生效。 这带来两个实用推论。其一，管道可以被反复迭代——for 循环跑完一轮，再迭代一次又从源头开始（shuffle 默认每轮重新洗牌，见下）。其二，在声明链上追加变换的顺序，就是数据流经各工位的顺序，顺序错了语义就错。

# 观察 shuffle 每轮重洗：迭代两轮，首批内容不同
first_round = [float(y[0]) for x, y in raw.shuffle(100).batch(1).take(3)]
second_round = [float(y[0]) for x, y in raw.shuffle(100).batch(1).take(3)]
print(first_round[:3])
print(second_round[:3])
# 两次输出不同——reshuffle_each_iteration 默认开启
# 验证集管道要保持确定性，用 seed=42 固定
​

三个基本变换的语义与时序

shuffle、batch、repeat 的排列组合决定训练的正确性。语义分别是：shuffle 在缓冲区里随机抽样打乱顺序；batch 把相邻元素堆成批次；repeat 把元素流复制多轮。顺序上的两条门道：

第一，shuffle 要在 batch 之前。先 batch 再 shuffle 只会打乱批次之间的顺序，每个批次内部的样本构成不变，打乱效果大打折扣。第二，batch 要在 repeat 之前。先 repeat 再 batch，一个 epoch 的边界会跨批次错开，且最后一个不满批次的行为难以控制。

# 正确顺序的标准范式
ds = (raw
      .shuffle(buffer_size=20640)     # 缓冲区最好接近数据集大小
      .batch(64)
      .repeat()                        # 无限流，交给 fit 的 steps_per_epoch 掐
      .prefetch(tf.data.AUTOTUNE))
print("pipeline built")
# 输出：pipeline built
# repeat() 无参调用是无限重复，
# 配合 model.fit(..., steps_per_epoch=323) 每轮掐满一遍数据

# buffer_size 的含义：不是"洗牌整段数据"，而是滑动窗口大小
small = raw.shuffle(buffer_size=10)    # 只在最近 10 条里随机抽
# 数据量小时把缓冲区开到全量；数据量太大时，
# 至少要大于"同标签样本的平均聚集跨度"，否则局部扎堆
​

观测管道而不惊动它

调试管道需要"看一眼产出"，但全量迭代违背惰性初衷。工具是 take 与 as_numpy_iterator：

sample = raw.shuffle(500).batch(4).take(1)
for xb, yb in sample.as_numpy_iterator():    # 转回 NumPy 便于观察
    print(xb[:2])
    print(yb[:2])
# 输出示例（随机洗牌，数值不定）：
# [[ 4.42   28.     5.34  ...]
#  [ 2.17   33.4    4.12  ...]]
# [3.72 1.85]

# 常用探针三件套
print(tf.data.Dataset.cardinality(raw.batch(64)).numpy())
# 输出：323 —— 数据集大小 20640 除以批次 64，Cardinality 是计划阶段的元素计数
​

⚠️ 常见坑：把管道在 for 循环外创建一次、循环里反复 fit 是对的；但给验证集管道也加无参 repeat，fit 的 validation 会永远算不完。repeat 只给训练管道用，且配 steps_per_epoch。

💡 关键直觉：看到 Dataset 代码，把它读成一句"计划句子"——从某处取元素、依次做这几件事、成批地交出来。计划先行、执行靠后，管道的一切怪现象都源于忘了"现在还在改计划，没有开机"。

本节要点回顾

• 惰性求值：变换是声明，迭代才执行；打印 Dataset 能看到嵌套计划结构。
• 标准范式：shuffle、batch、repeat、prefetch 的固定顺序与各自语义。
• shuffle 在 batch 前：先打乱再成批才有效；buffer_size 是滑动窗口。
• repeat 在 batch 后：无参 repeat 是无限流，配 steps_per_epoch 使用。
• 观测工具：take 限量迭代，as_numpy_iterator 转回 NumPy，cardinality 数元素。

下节把"数据从哪来"拆成三种上料方式细讲。