整个数据API都围绕着数据集的概念。通常，使用的是逐步从磁盘中读取数据的数据集，但为了简单起见，使用

tf.data.Dataset.from_tensor_slices()在RAM中完全创建一个数据集合

import keras.activations

import tensorflow as tf

X = tf.range(10)

dataset = tf.data.Dataset.from_tensor_slices(X)

dataset

1.

2.

3.

4.

5.

6.

<TensorSliceDataset shapes: (), types: tf.int32>

1.

from_tensor_slices()函数采用一个张量并创建一个tf.data.Dataset，其元素都是X的去切片（沿第一个维度），

因此此数据集包含10个元素：张量0,1,2，···，9。在这种情况下，如果使用tf.data.Dataset.range(10)，则将获得相同的数据集

可以通过for循环简单的遍历数据集的元素

for item in dataset:

    print(item)

1.

2.

tf.Tensor(0, shape=(), dtype=int32)

tf.Tensor(1, shape=(), dtype=int32)

tf.Tensor(2, shape=(), dtype=int32)

tf.Tensor(3, shape=(), dtype=int32)

tf.Tensor(4, shape=(), dtype=int32)

tf.Tensor(5, shape=(), dtype=int32)

tf.Tensor(6, shape=(), dtype=int32)

tf.Tensor(7, shape=(), dtype=int32)

tf.Tensor(8, shape=(), dtype=int32)

tf.Tensor(9, shape=(), dtype=int32)

1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

链式转换

有了数据集之后，可以通过调用其转换方法对其应用各种转换。每个方法都返回一个新的数据集，因此可以进行链式转换

dataset = dataset.repeat(3).batch(7)

for item in dataset:

    print(item)

1.

2.

3.

tf.Tensor([0 1 2 3 4 5 6], shape=(7,), dtype=int32)

tf.Tensor([7 8 9 0 1 2 3], shape=(7,), dtype=int32)

tf.Tensor([4 5 6 7 8 9 0], shape=(7,), dtype=int32)

tf.Tensor([1 2 3 4 5 6 7], shape=(7,), dtype=int32)

tf.Tensor([8 9], shape=(2,), dtype=int32)

1.

2.

3.

4.

5.

首先在原始数据集上调用repeat()方法，它会返回一个新的数据集，该数据集将重复原始数据集的元素三次。然后，batch()方法

最后输出一个大小为2而不是7的最终批次，但是如果希望它删除最终批次，可以使用drop_reminder=True调用它，是所有批次具有完全相同的大小

# 通过调用map()方法来变换元素。例如，创建一个新的数据集，其中所有元素均是原来的两倍：

dataset = dataset.map(lambda x: x ** 2)

# map()方法将转换应用于每个元素，但是apply()方法将转换应用于整个数据集。

# 将unbatch()方法应用于数据集

dataset = dataset.apply(tf.data.experimental.unbatch())

# 也可以使用filter()方法简单地过滤数据集：

dataset = dataset.filter(lambda x: x < 10)

# 查看数据集中的一些元素，可以使用take()方法

for item in dataset.take(3):

    print(item)

1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

乱序数据

当训练集中的实例相互独立且分布均匀时，梯度下降效果佳。确保这一点的简单方式是使用shuffle()方法对

实例进行乱序。它会创建一个新的数据集，该数据集首先将源数据集的第一项填充到缓冲区。然后无论任何时候要求

提供一个数据，它都会从缓冲区中随机取出一个元素，并用源数据集中的新元素替换它，直到完全遍历源数据集为止。

它将继续从缓冲区随机抽取元素直到其为空。这就必须要指定缓冲区的大小，重要的是使其足够大，

否则乱序不会非常有效。不要超出你有的RAM区数量，即使拥有足够的RAM，也不需要超出数据集的大小。

如果每次运行程序都想要相同的随机顺序，可以提供种子。例如，一下代码创建并显示一个包含整数0到9的

数据集，重复3次，使用大小为5的缓冲区和42的随机种子进行乱序，并以7的批次大小进行批处理

dataset = tf.data.Dataset.range(10).repeat(3)

dataset = dataset.shuffle(buffer_size=5, seed=42).batch(7)

for item in dataset:

    print(item)

1.

2.

3.

4.

tf.Tensor([0 2 3 6 7 9 4], shape=(7,), dtype=int64)

tf.Tensor([5 0 1 1 8 6 5], shape=(7,), dtype=int64)

tf.Tensor([4 8 7 1 2 3 0], shape=(7,), dtype=int64)

tf.Tensor([5 4 2 7 8 9 9], shape=(7,), dtype=int64)

tf.Tensor([3 6], shape=(2,), dtype=int64)

1.

2.

3.

4.

5.

# 交织来自多个文件的行

filepath_dataset = tf.data.Dataset.list_files(train_filepaths, seed=42)

# 默认情况下，list_files()函数返回一个乱序的文件路径的数据集。通常这是一件后世

# 如果出于某些原因不希望这么做，则可以设置shuffle=False

# 接下来可以调用interleave()方法一次读取个文件并交织它们的行（使用skip()方法跳过

# 每个文件的第一行，即标题行）：

n_readers = 5

dataset = filepath_dataset.interleave(

    lambda filepath: tf.data.TextLineDataset(filepath).skip(1),

    cycle_length=n_readers)

)

1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

interleave()方法将创建一个数据集，该数据集将从filepath_dataset中拉出5个文件路径,

对于每个路径，它将调用你为其提供的函数（在此示例中为lambda）来创建新的数据集

（在此示例中为TextLineDataset）。为了清楚起见，在此阶段总共有7个数据集：文件路径数据集、

交织数据集和由交织数据集在内部创建的5个TextLineDataset。当遍历交织数据集时，它将循环

遍历这5个TextLineDatasets，每次读取一行，直到所有数据集都读出位置。然后它从

filepath_dataset获取剩下的5个文件路径，并以相同的方式对它们进行交织，以此类推，直到读完文件路径。

预处理数据

import tensorflow as tf

X_mean,X_std=[...]

n_inputs=8

def preprocess(line):

    defs=[.0]*n_inputs+[tf.constant([],dtype=tf.float32)]

    fields=tf.io.decode_csv(line,record_defaults=defs)

    x=tf.stack(fields[:-1])

    y=tf.stack(fields[-1:])

    return (x-X_mean)/X_std,y

1.

2.

3.

4.

5.

6.

7.

8.

9.

首先，代码假设已经预先计算了训练集中每个特征的均值和标准差。X_mean和X_std是一维

张量（或NumPy数组），其中包含8个浮点数，每个输入特征一个。

preprocess()函数使用一行CSV内容并开始对其进行解析。为此它使用tf.io.decode_csv

()函数，该函数带有两个参数：第一个是要解析的行，第二个是一个包含CSV文件中每一行的默认值的数组。

这个数组要告诉TensorFlow每列的默认值，并且告诉列数及其类型。在此实例中，告诉所有特征数列

都是浮点数，确实砂纸应默认为0，还提供了一个类型为tf.float32的空数组作为最后一行（目标值）

的默认值：该数组告诉TensorFlow该列包含浮点数，但没有默认值，因此如果遇到缺失值，它会引发异常

encode_csv()函数返回标量张量的列表（每列一个），但是需要返回一维张量数组。

因此在出最后一个（目标值）之外的所有张量上调用tf.stack()：这会将这些张量堆叠到一维度组中。然后对目标值执行相同的操作（这使其成为具有单个值的一维张量数组，而不是标量张量）

最后通过减去特征均值然后除以特征标准差来缩放输入特征，然后返回一个包含已缩放特征值和目标值的元组

合并在一起

为了使代码可重用，将目前为止的所有内容放到一个小的辅助函数中：它将创建并返回一个数据集，该数据集有效地从多个CSV文件

中加载加州住房数据，对其进行预处理、随机乱序，可重复选择，并进行批处理

n_inputs = 8

def preprocess(line):

    defs = [.0] * n_inputs + [tf.constant([], dtype=tf.float32)]

    fields = tf.io.decode_csv(line, record_defaults=defs)

    x = tf.stack(fields[:-1])

    y = tf.stack(fields[-1:])

    return (x - X_mean) / X_std, y

def csv_reader_dataset(filepaths, repeat=1, n_readers=5, n_read_threads=None, shuffle_buffer_size=10000,

                       n_parse_threads=5, batch_size=32):

    dataset = tf.data.Dataset.list_files(filepaths)

    dataset = dataset.interleave(

        lambda filepath: tf.data.TextLineDataset(filepaths).skip(1),

        cycle_length=n_readers, num_parallel_calls=n_read_threads

    )

    dataset = dataset.map(preprocess, number_parallel_calls=n_parse_threads)

    dataset = dataset.shuffle(shuffle_buffer_size).repeat(repeat)

    return dataset.batch(batch_size).prefetch(1)

1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

预取

通过最后调用prefetch(1)，创建一个数据集，该数据集将尽最大可能总是提前准备一个批次。换句话说，训练算法正在处理

一个批次时，数据集已经并行工作以准备下一批次了，这可以显著提高性能。如果确保加载和预处理是多线程的（通过

在调用interleave()和map()时设置num_parallel_calls），可以在CPU上利用多个内核，希望准备一个批次的时间

比在GPU上执行一个训练步骤的时间要短一些：这样，GPU将达到几乎100%的利用率（从CPU到GPU的数据传输时间除外），

并且训练会运行得更快

和tf.keras一起使用数据集

现在可以用csv_reader_dataset()函数为训练集创建数据集。

train_set = csv_reader_dataset(train_filepaths)

valid_set = csv_reader_dataset(valid_filepaths)

test_set = csv_reader_dataset(test_filepaths)

1.

2.

3.

现在可以使用这些数据集简单的构建Keras模型。将训练数据集和验证数据集传递给fit()方法，而不是X_train,y_trian,X_valid和

y_valid

model = keras.Sequential([...])

model.compile([...])

model.fit(train_set, epochs=10, validation_data=valid_set)

# 类似的可以将数据集传递给valuate()方法和predict()方法

model.evaluate(test_set)

new_set = test_set.take(3).map(lambda X, y X)

model.predict(new_set)

# 如果要构建自己的自定义训练循环，可以非常自然地遍历训练集：

for X_batch, y_batch in train_set:

    [...]

# 实际上甚至可以创建执行整个训练循环的TF函数：

@tf.function

def train(model, optimizer, loss_fn, n_epochs):

    train_set = csv_reader_dataset(train_filepaths, repeat=n_epochs, [...])

    for X_batch, y_batch in train_set:

        with tf.GradientTape() as tape:

            y_pred = model(X_batch)

            main_loss = tf.reduce_mean(loss_fn(y_batch, y_pred))

            loss = tf.add_n([main_loss] + model.losses)

        grads = tape.gradient(loss, model.trainable_variables)

        optimizer.apply_graidnets(zip(grads, model.trainable_variables))

-----------------------------------