链接转换

一次您有一个数据集，您可以通过调用其转换方法对其应用各种转换。每个方法都返回一个新的数据集，所以你可以像这样链接转换（这个链如图 13-1 所示）：

>>> dataset = dataset.repeat(3).batch(7)
>>> for item in dataset:
...     print(item)
...
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)

在本例中，我们首先在原始数据集上调用该repeat()方法，它返回一个新数据集，该数据集将重复原始数据集的项目 3 次。当然，这不会将内存中的数据全部复制三遍！（如果你在没有参数的情况下调用这个方法，新数据集将永远重复源数据集，所以迭代数据集的代码必须决定何时停止。）然后我们batch()在这个新数据集上调用该方法，再一次创建一个新的数据集。这会将前一个数据集的项目分组为七个项目。最后，我们迭代这个最终数据集的项目。如您所见，该batch()方法必须输出最后一批大小为 2 而不是 7 的批次，但您可以使用drop_remainder=True如果您希望它删除最后一批，以便所有批次具有完全相同的大小。

您还可以通过调用该map()方法来转换项目。例如，这会创建一个所有项目都翻倍的新数据集：

>>> dataset = dataset.map(lambda x: x * 2) # Items: [0,2,4,6,8,10,12]

此函数是您将调用以对数据应用任何预处理的函数。有时这将包括可能非常密集的计算，例如重塑或旋转图像，因此您通常希望生成多个线程以加快速度：就像设置num_parallel_calls参数一样简单。请注意，您传递给map()方法的函数必须可转换为 TF 函数（参见第 12 章）。

虽然该map()方法将转换应用于每个项目，但该apply()方法将转换应用于整个数据集。例如，以下代码将该unbatch()函数应用于数据集1。新数据集中的每个项目都将是一个单整数张量，而不是一组七个整数：

>>> dataset = dataset.apply(tf.data.experimental.unbatch()) # Items: 0,2,4,...

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

>>> dataset = dataset.filter(lambda x: x < 10) # Items: 0 2 4 6 8 0 2 4 6...

您通常只想查看数据集中的几个项目。您可以使用以下take()方法：

>>> for item in dataset.take(3):
...     print(item)
...
tf.Tensor(0, shape=(), dtype=int64)
tf.Tensor(2, shape=(), dtype=int64)
tf.Tensor(4, shape=(), dtype=int64)

洗牌数据（Shuffling the Data）

作为你知道，当训练集中的实例独立且同分布时，梯度下降效果最好（参见第 4 章）。确保这一点的一种简单方法是使用该shuffle()方法对实例进行洗牌。它将创建一个新数据集，该数据集将首先用源数据集的第一项填充缓冲区。然后，每当它被要求提供一个项目时，它都会从缓冲区中随机抽取一个，并用源数据集中的一个新项目替换它，直到它完全遍历源数据集。此时它会继续从缓冲区中随机拉出项目，直到它为空。您必须指定缓冲区大小，并且使其足够大很重要，否则洗牌不会很有效。2只是不要超过你拥有的 RAM 数量，即使你有很多，也没有必要超出数据集的大小。如果每次运行程序时都想要相同的随机顺序，可以提供随机种子。例如，以下代码创建并显示一个包含整数 0 到 9 的数据集，重复 3 次，使用大小为 5 的缓冲区和随机种子 42 进行混洗，并使用批量大小为 7 进行批处理：

>>> dataset = tf.data.Dataset.range(10).repeat(3) # 0 to 9, three times
>>> dataset = dataset.shuffle(buffer_size=5, seed=42).batch(7)
>>> for item in dataset:
...     print(item)
...
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)

小费

如果你调用repeat()一个打乱的数据集，默认情况下它会在每次迭代时生成一个新的顺序。这通常是一个好主意，但如果您希望在每次迭代中重用相同的顺序（例如，用于测试或调试），您可以设置reshuffle_each_iteration=False.

为了对于一个不适合内存的大型数据集，这种简单的洗牌缓冲区方法可能还不够，因为与数据集相比，缓冲区会很小。一种解决方案是打乱源数据本身（例如，在 Linux 上，您可以使用shuf命令）。这肯定会改善洗牌很多！即使源数据被打乱了，您通常也希望将其打乱一些，否则在每个时期都会重复相同的顺序，并且模型最终可能会出现偏差（例如，由于偶然出现的一些虚假模式源数据的顺序）。为了进一步打乱实例，一种常见的方法是将源数据拆分为多个文件，然后在训练期间以随机顺序读取它们。但是，位于同一文件中的实例仍然会彼此靠近。为避免这种情况，您可以随机选择多个文件并同时读取它们，交错它们的记录。然后最重要的是，您可以使用shuffle()方法。如果所有这些听起来都需要大量工作，请不要担心：Data API 只需几行代码即可完成所有这些工作。让我们看看如何做到这一点。

MedInc,HouseAge,AveRooms,AveBedrms,Popul,AveOccup,Lat,Long,MedianHouseValue 3.5214,15.0,3.0499,1.1065,1447.0,1.6059,37.63,-122.43,1.442 5.3275,5.0,6.4900,0.9910,3464.0,3.4433,33.69,-117.39,1.687 3.1,29.0,7.5423,1.5915,1328.0,2.2508,38.44,-122.98,1.621 [...]

>>> train_filepaths['datasets/housing/my_train_00.csv', 'datasets/housing/my_train_01.csv',...]

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

n_readers = 5
dataset = filepath_dataset.interleave(
    lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
    cycle_length=n_readers)

>>> for line in dataset.take(5):
...     print(line.numpy())
...
b'4.2083,44.0,5.3232,0.9171,846.0,2.3370,37.47,-122.2,2.782'
b'4.1812,52.0,5.7013,0.9965,692.0,2.4027,33.73,-118.31,3.215'
b'3.6875,44.0,4.5244,0.9930,457.0,3.1958,34.04,-118.15,1.625'
b'3.3456,37.0,4.5140,0.9084,458.0,3.2253,36.67,-121.7,2.526'
b'3.5214,15.0,3.0499,1.1065,1447.0,1.6059,37.63,-122.43,1.442'

预处理数据

让我们实现一个小函数来执行这个预处理：

X_mean, X_std = [...] # mean and scale of each feature in the training set
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

让我们看一下这段代码：

• 首先，代码假设我们已经预先计算了训练集中每个特征的均值和标准差。X_mean并且X_std只是包含八个浮点数的一维张量（或 NumPy 数组），每个输入特征一个。

• 该preprocess()函数采用一个 CSV 行并从解析它开始。为此，它使用该tf.io.decode_csv()函数，该函数接受两个参数：第一个是要解析的行，第二个是包含 CSV 文件中每一列的默认值的数组。这个数组不仅告诉 TensorFlow 每一列的默认值，还告诉 TensorFlow 的列数和它们的类型。在这个例子中，我们告诉它所有特征列都是浮点数并且缺失值应该默认为 0，但是我们提供了一个空类型的数组tf.float32作为最后一列（目标）的默认值：该数组告诉 TensorFlow 该列包含浮点数，但没有默认值，因此如果遇到缺失值，它将引发异常。

• 该decode_csv()函数返回一个标量张量列表（每列一个），但我们需要返回一维张量数组。所以我们调用tf.stack()除了最后一个（目标）之外的所有张量：这会将这些张量堆叠成一维数组。然后我们对目标值做同样的事情（这使它成为一个具有单个值的一维张量数组，而不是一个标量张量）。

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

让我们测试一下这个预处理功能：

>>> preprocess(b'4.2083,44.0,5.3232,0.9171,846.0,2.3370,37.47,-122.2,2.782')
(<tf.Tensor: id=6227, shape=(8,), dtype=float32, numpy=
 array([ 0.16579159,  1.216324  , -0.05204564, -0.39215982, -0.5277444 ,
        -0.2633488 ,  0.8543046 , -1.3072058 ], dtype=float32)>,
 <tf.Tensor: [...], numpy=array([2.782], dtype=float32)>)

看起来不错！我们现在可以将该函数应用于数据集。

把所有东西放在一起

至使代码可重用，让我们将到目前为止讨论的所有内容放在一个小型辅助函数中：它将创建并返回一个数据集，该数据集将从多个 CSV 文件中有效加载加州住房数据，对其进行预处理、随机播放、选择性地重复它，以及批处理它（见图 13-2）：

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(filepath).skip(1),
        cycle_length=n_readers, num_parallel_calls=n_read_threads)
    dataset = dataset.shuffle(shuffle_buffer_size).repeat(repeat)
    dataset = dataset.map(preprocess, num_parallel_calls=n_parse_threads)
    return dataset.batch(batch_size).prefetch(1)

这段代码中的所有内容都应该是有意义的，除了最后一行 ( prefetch(1))，这对性能很重要。

预取

经过最后调用prefetch(1)，我们正在创建一个数据集，该数据集将尽最大努力始终领先一批。3换句话说，当我们的训练算法在一个批次上工作时，数据集已经在并行工作以准备下一个批次（例如，从磁盘读取数据并对其进行预处理）。这可以显着提高性能，如图 13-3 所示。如果我们还确保加载和预处理是多线程的（通过num_parallel_calls在调用interleave()和时设置map()），我们可以利用 CPU 上的多个内核，并希望使准备一批数据比在 GPU 上运行训练步骤更短：这样 GPU 将几乎 100% 被利用（除了从 CPU 到 GPU 的数据传输时间4)，训练会运行得更快。

如果数据集足够小以适合内存，则可以通过使用数据集的cache()方法将其内容缓存到 RAM 来显着加快训练速度。您通常应该在加载和预处理数据之后，但在改组、重复、批处理和预取之前执行此操作。这样，每个实例只会被读取和预处理一次（而不是每个 epoch 一次），但数据在每个 epoch 仍然会以不同的方式洗牌，并且仍然会提前准备下一批。

您现在知道如何构建高效的输入管道来加载和预处理来自多个文本文件的数据。我们已经讨论了最常见的数据集方法，但您可能还想了解更多：concatenate()、zip()、window()、reduce()、shard()、flat_map()和padded_batch()。还有几个类方法：from_generator()和from_tensors()，它们分别从 Python 生成器或张量列表创建新数据集。请查看 API 文档以获取更多详细信息。另请注意，在 中提供了一些实验性功能tf.data.experimental，其中许多可能会在未来的版本中成为核心 API（例如，检查CsvDataset类和make_csv_dataset()方法，该方法负责推断每列的类型）。

将数据集与 tf.keras 一起使用

现在我们可以使用该csv_reader_dataset()函数为训练集创建数据集。请注意，我们不需要重复它，因为这将由 tf.keras 处理。我们还为验证集和测试集创建数据集：

train_set = csv_reader_dataset(train_filepaths)
valid_set = csv_reader_dataset(valid_filepaths)
test_set = csv_reader_dataset(test_filepaths)

现在我们可以使用这些数据集简单地构建和训练 Keras 模型。5我们需要做的就是将训练和验证数据集传递给fit()方法，而不是X_train, y_train、X_valid和y_valid：6

model = keras.models.Sequential([...])
model.compile([...])
model.fit(train_set, epochs=10, validation_data=valid_set)

同样，我们可以将数据集传递给evaluate()andpredict()方法：

model.evaluate(test_set)
new_set = test_set.take(3).map(lambda X, y: X) # pretend we have 3 new instances
model.predict(new_set) # a dataset containing new instances

与其他集合不同，new_set通常不包含标签（如果包含，Keras 将忽略它们）。请注意，在所有这些情况下，如果需要，您仍然可以使用 NumPy 数组而不是数据集（当然，它们需要先被加载和预处理）。

如果你想构建自己的自定义训练循环（如第 12 章所述），你可以很自然地迭代训练集：

for X_batch, y_batch in train_set:
    [...] # perform one Gradient Descent step

事实上，甚至可以创建一个执行整个训练循环的 TF 函数（参见第 12 章）：

@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_gradients(zip(grads, model.trainable_variables))

恭喜，您现在知道如何构建强大的使用数据 API 的输入管道！但是，到目前为止，我们使用的 CSV 文件很常见、简单、方便，但效率并不高，而且不能很好地支持大型或复杂的数据结构（例如图像或音频）。所以让我们看看如何改用 TFRecords。

小费

默认情况下，aTFRecordDataset会逐个读取文件，但您可以通过设置使其并行读取多个文件并交错记录num_parallel_reads。或者，您可以通过使用list_files()和获得相同的结果，interleave()就像我们之前所做的那样读取多个 CSV 文件。

压缩的 TFRecord 文件

它有时对压缩 TFRecord 文件很有用，尤其是当它们需要通过网络连接加载时。您可以通过设置参数来创建压缩的 TFRecord 文件options：

options = tf.io.TFRecordOptions(compression_type="GZIP")
with tf.io.TFRecordWriter("my_compressed.tfrecord", options) as f:
  [...]

读取压缩的 TFRecord 文件时，需要指定压缩类型：

dataset = tf.data.TFRecordDataset(["my_compressed.tfrecord"],
                                  compression_type="GZIP")

协议缓冲区简介

甚至尽管每条记录都可以使用您想要的任何二进制格式，但 TFRecord 文件通常包含序列化的协议缓冲区（也称为protobufs）。这是一种可移植的、可扩展的、高效的二进制格式，早在 2001 年由 Google 开发，并于 2008 年开源；protobufs 现在被广泛使用，特别是在gRPC中，谷歌的远程过程调用系统。它们是使用如下所示的简单语言定义的：

syntax = "proto3";
message Person {
  string name = 1;
  int32 id = 2;
  repeated string email = 3;
}

这个定义说我们使用的是 protobuf 格式的版本 3，它指定每个Person对象7可以（可选地）有一个nametype string，一个idtype int32，以及零个或多个email字段，每个 type string。数字1、2和3是字段标识符：它们将用于每条记录的二进制表示。一旦你在.proto文件中有一个定义，你就可以编译它。这需要protocprotobuf 编译器以 Python（或其他语言）生成访问类。请注意，我们将使用的 protobuf 定义已经为您编译好了，它们的 Python 类是 TensorFlow 的一部分，因此您不需要使用protoc. 你只需要知道如何在 Python 中使用 protobuf 访问类。为了说明基础知识，让我们看一个使用为Personprotobuf 生成的访问类的简单示例（代码在注释中解释）：

>>> from person_pb2 import Person  # import the generated access class
>>> person = Person(name="Al", id=123, email=["a@b.com"])  # create a Person
>>> print(person)  # display the Person
name: "Al"
id: 123
email: "a@b.com"
>>> person.name  # read a field
"Al"
>>> person.name = "Alice"  # modify a field
>>> person.email[0]  # repeated fields can be accessed like arrays
"a@b.com"
>>> person.email.append("c@d.com")  # add an email address
>>> s = person.SerializeToString()  # serialize the object to a byte string
>>> s
b'\n\x05Alice\x10{\x1a\x07a@b.com\x1a\x07c@d.com'
>>> person2 = Person()  # create a new Person
>>> person2.ParseFromString(s)  # parse the byte string (27 bytes long)
27
>>> person == person2  # now they are equal
True

简而言之，我们导入由Person生成的类protoc，我们创建一个实例并使用它，将其可视化并读写一些字段，然后使用该SerializeToString()方法对其进行序列化。这是准备好通过网络保存或传输的二进制数据。当读取或接收这个二进制数据时，我们可以使用该ParseFromString()方法对其进行解析，得到一个被序列化的对象的副本。8

我们可以将序列化的Person对象保存到 TFRecord 文件中，然后我们可以加载并解析它：一切都会正常工作。但是，SerializeToString()并且ParseFromString()不是 TensorFlow 操作（此代码中的其他操作也不是），因此它们不能包含在 TensorFlow 函数中（除非将它们包装在一个tf.py_function()操作中，这会使代码更慢且更不便携，正如我们所见在第 12 章中）。幸运的是，TensorFlow 确实包含了特殊的 protobuf 定义，它为其提供了解析操作。

TensorFlow Protobufs

这TFRecord 文件中通常使用的主要 protobuf 是Exampleprotobuf，它表示数据集中的一个实例。它包含一个命名特征列表，其中每个特征可以是字节字符串列表、浮点列表或整数列表。这是 protobuf 的定义：

syntax = "proto3";
message BytesList { repeated bytes value = 1; }
message FloatList { repeated float value = 1 [packed = true]; }
message Int64List { repeated int64 value = 1 [packed = true]; }
message Feature {
    oneof kind {
        BytesList bytes_list = 1;
        FloatList float_list = 2;
        Int64List int64_list = 3;
    }
};
message Features { map<string, Feature> feature = 1; };
message Example { Features features = 1; };

BytesList、FloatList和的定义Int64List很简单。请注意，[packed = true]它用于重复的数字字段，以实现更有效的编码。A Feature包含 a BytesList、 aFloatList或 an Int64List。A Features(with an s) 包含一个将特征名称映射到相应特征值的字典。最后，anExample只包含一个Features对象。9以下是如何创建一个tf.train.Example代表与之前相同的人并将其写入 TFRecord 文件的方法：

from tensorflow.train import BytesList, FloatList, Int64List
from tensorflow.train import Feature, Features, Example

person_example = Example(
    features=Features(
        feature={
            "name": Feature(bytes_list=BytesList(value=[b"Alice"])),
            "id": Feature(int64_list=Int64List(value=[123])),
            "emails": Feature(bytes_list=BytesList(value=[b"a@b.com",
                                                          b"c@d.com"]))
        }))

代码有点冗长和重复，但它相当简单（你可以很容易地将它包装在一个小的帮助函数中）。现在我们有了一个Exampleprotobuf，我们可以通过调用它的SerializeToString()方法来序列化它，然后将结果数据写入一个 TFRecord 文件：

with tf.io.TFRecordWriter("my_contacts.tfrecord") as f:
    f.write(person_example.SerializeToString())

通常你会写不止一个Example！通常，您将创建一个从当前格式（例如 CSV 文件）读取的转换脚本，Example为每个实例创建一个 protobuf，将它们序列化，并将它们保存到多个 TFRecord 文件中，理想情况下在此过程中将它们打乱。这需要一些工作，所以再次确保它确实是必要的（也许您的管道可以正常处理 CSV 文件）。

现在我们有了一个包含 serialized 的漂亮 TFRecord 文件Example，让我们尝试加载它。

加载和解析示例

至加载序列化的Exampleprotobufs，我们将tf.data.TFRecordDataset再次使用 a，我们将Example使用tf.io.parse_single_example(). 这是一个 TensorFlow 操作，因此它可以包含在 TF 函数中。它至少需要两个参数：一个包含序列化数据的字符串标量张量，以及每个特征的描述。描述是一个字典，它将每个特征名称映射到tf.io.FixedLenFeature指示特征的形状、类型和默认值的描述符，或tf.io.VarLenFeature仅指示类型的描述符（如果特征列表的长度可能不同，例如"emails"特征）。

下面的代码定义了一个描述字典，然后它遍历TFRecordDataset并解析了Example这个数据集包含的序列化 protobuf：

feature_description = {
    "name": tf.io.FixedLenFeature([], tf.string, default_value=""),
    "id": tf.io.FixedLenFeature([], tf.int64, default_value=0),
    "emails": tf.io.VarLenFeature(tf.string),
}

for serialized_example in tf.data.TFRecordDataset(["my_contacts.tfrecord"]):
    parsed_example = tf.io.parse_single_example(serialized_example,
                                                feature_description)

固定长度的特征被解析为常规张量，而可变长度的特征被解析为稀疏张量。您可以使用 将稀疏张量转换为密集张量tf.sparse.to_dense()，但在这种情况下，仅访问其值会更简单：

>>> tf.sparse.to_dense(parsed_example["emails"], default_value=b"")
<tf.Tensor: [...] dtype=string, numpy=array([b'a@b.com', b'c@d.com'], [...])>
>>> parsed_example["emails"].values
<tf.Tensor: [...] dtype=string, numpy=array([b'a@b.com', b'c@d.com'], [...])>

ABytesList可以包含您想要的任何二进制数据，包括任何序列化对象。例如，您可以使用tf.io.encode_jpeg()JPEG 格式对图像进行编码，并将此二进制数据放入BytesList. 稍后，当您的代码读取 TFRecord 时，它会从解析 开始Example，然后需要调用tf.io.decode_jpeg()解析数据并获取原始图像（或者您可以使用tf.io.decode_image()，它可以解码任何 BMP、GIF、JPEG 或 PNG 图像）。您还可以BytesList通过使用序列化张量tf.io.serialize_tensor()然后将生成的字节字符串放入BytesList特征中来存储您想要的任何张量。稍后，当您解析 TFRecord 时，您可以使用tf.io.parse_tensor().

tf.io.parse_single_example()您可能希望使用 逐批解析它们，而不是使用 逐个解析示例tf.io.parse_example()：

dataset = tf.data.TFRecordDataset(["my_contacts.tfrecord"]).batch(10)
for serialized_examples in dataset:
    parsed_examples = tf.io.parse_example(serialized_examples,
                                          feature_description)

如您所见，Exampleprotobuf 可能足以满足大多数用例。但是，在处理列表列表时使用它可能有点麻烦。例如，假设您要对文本文档进行分类。每个文档可以表示为句子列表，其中每个句子表示为单词列表。也许每个文档也有一个评论列表，其中每个评论都表示为一个单词列表。可能还有一些上下文数据，例如文档的作者、标题和发布日期。TensorFlow 的SequenceExampleprotobuf 就是为此类用例而设计的。

使用 SequenceExample Protobuf 处理列表的列表

这里是SequenceExampleprotobuf的定义：

message FeatureList { repeated Feature feature = 1; };
message FeatureLists { map<string, FeatureList> feature_list = 1; };
message SequenceExample {
    Features context = 1;
    FeatureLists feature_lists = 2;
};

ASequenceExample包含一个Features用于上下文数据的FeatureLists对象和一个包含一个或多个命名FeatureList对象（例如，一个FeatureList命名对象"content"和另一个命名对象"comments"）的对象。每个都FeatureList包含一个Feature对象列表，每个对象可能是一个字节字符串列表、一个 64 位整数列表或一个浮点数列表（在这个例子中，每个Feature都代表一个句子或评论，可能形式为单词标识符列表）。构建SequenceExample、序列化和解析它与构建、序列化和解析 an 类似Example，但您必须使用它tf.io.parse_single_sequence_example()来解析单个SequenceExample或tf.io.parse_sequence_example()解析一批。这两个函数都返回一个包含上下文特征（作为字典）和特征列表（也作为字典）的元组。如果特征列表包含不同大小的序列（如前面的示例中所示），您可能希望使用以下方法将它们转换为不规则张量tf.RaggedTensor.from_sparse()（请参阅笔记本以获取完整代码）：

parsed_context, parsed_feature_lists = tf.io.parse_single_sequence_example(
    serialized_sequence_example, context_feature_descriptions,
    sequence_feature_descriptions)
parsed_content = tf.RaggedTensor.from_sparse(parsed_feature_lists["content"])

既然您知道如何有效地存储、加载和解析数据，下一步就是准备数据，以便将其馈送到神经网络。

使用 One-Hot 向量对分类特征进行编码

考虑我们在第 2 章ocean_proximity探讨的加利福尼亚住房数据集中的特征：它是一个具有五个可能值的分类特征：、、、和。我们需要先对这个特征进行编码，然后再将其输入神经网络。由于类别很少，我们可以使用 one-hot 编码。为此，我们首先需要将每个类别映射到其索引（0 到 4），这可以使用查找表来完成："<1H OCEAN""INLAND""NEAR OCEAN""NEAR BAY""ISLAND"

vocab = ["<1H OCEAN", "INLAND", "NEAR OCEAN", "NEAR BAY", "ISLAND"]
indices = tf.range(len(vocab), dtype=tf.int64)
table_init = tf.lookup.KeyValueTensorInitializer(vocab, indices)
num_oov_buckets = 2
table = tf.lookup.StaticVocabularyTable(table_init, num_oov_buckets)

让我们看一下这段代码：

• 我们首先定义词汇表：这是所有可能类别的列表。

• 然后我们创建一个具有相应索引（0 到 4）的张量。

• 接下来，我们为查找表创建一个初始化程序，将类别列表及其对应的索引传递给它。在这个例子中，我们已经有了这些数据，所以我们使用KeyValueTensorInitializer; 但如果类别在文本文件中列出（每行一个类别），我们将使用 aTextFileInitializer代替。

• 在最后两行我们创建查找表，为其提供初始化程序并指定词汇表外（oov）桶的数量。如果我们查找词汇表中不存在的类别，则查找表将计算该类别的哈希并使用它来将未知类别分配给 oov 桶之一。它们的索引在已知类别之后开始，因此在此示例中，两个 oov 存储桶的索引为 5 和 6。

为什么使用 oov 存储桶？好吧，如果类别的数量很大（例如，邮政编码、城市、单词、产品或用户）并且数据集也很大，或者它一直在变化，那么获取完整的类别列表可能并不方便。一种解决方案是基于数据样本（而不是整个训练集）定义词汇表，并为不在数据样本中的其他类别添加一些 oov 桶。您希望在训练期间找到的未知类别越多，您应该使用的 oov 存储桶就越多。确实，如果没有足够的oov桶，就会发生碰撞：不同的类别最终会在同一个桶中，因此神经网络将无法区分它们（至少不能基于这个特征）。

现在让我们使用查找表将一小批分类特征编码为 one-hot 向量：

>>> categories = tf.constant(["NEAR BAY", "DESERT", "INLAND", "INLAND"])
>>> cat_indices = table.lookup(categories)
>>> cat_indices
<tf.Tensor: id=514, shape=(4,), dtype=int64, numpy=array([3, 5, 1, 1])>
>>> cat_one_hot = tf.one_hot(cat_indices, depth=len(vocab) + num_oov_buckets)
>>> cat_one_hot
<tf.Tensor: id=524, shape=(4, 7), dtype=float32, numpy=
array([[0., 0., 0., 1., 0., 0., 0.],
       [0., 0., 0., 0., 0., 1., 0.],
       [0., 1., 0., 0., 0., 0., 0.],
       [0., 1., 0., 0., 0., 0., 0.]], dtype=float32)>

如您所见，"NEAR BAY"被映射到索引 3，未知类别"DESERT"被映射到两个 oov 存储桶之一（在索引 5 处），"INLAND"并被映射到索引 1，两次。然后我们习惯于tf.one_hot()对这些索引进行 one-hot 编码。请注意，我们必须告诉这个函数索引的总数，它等于词汇表大小加上 oov 桶的数量。现在您知道如何使用 TensorFlow 将分类特征编码为 one-hot 向量！

就像之前一样，将所有这些逻辑捆绑到一个不错的自包含类中并不难。它的adapt()方法将获取数据样本并提取其中包含的所有不同类别。它将创建一个查找表以将每个类别映射到其索引（包括使用 oov 存储桶的未知类别）。然后它的call()方法将使用查找表将输入类别映射到它们的索引。好吧，还有更多好消息：当您阅读本文时，Keras 可能会包含一个keras.layers.TextVectorization名为它在词汇表中的索引。您可以在模型的开头添加此层，然后添加adapt()call()Lambdatf.one_hot()如果要将这些索引转换为单热向量，则将应用该函数的层。

不过，这可能不是最好的解决方案。每个 one-hot 向量的大小是词汇长度加上 oov 桶的数量。当只有几个可能的类别时这很好，但如果词汇量很大，编码它们的效率要高得多改为使用嵌入。

使用嵌入对分类特征进行编码

一个嵌入是表示一个类别的可训练密集向量。默认情况下，嵌入是随机初始化的，例如，"NEAR BAY"类别最初可以由一个随机向量表示，例如[0.131, 0.890]，而"NEAR OCEAN"类别可能由另一个随机向量表示，例如[0.631, 0.791]。在此示例中，我们使用 2D 嵌入，但维数是您可以调整的超参数。由于这些嵌入是可训练的，它们会在训练过程中逐渐提高；并且由于它们代表了相当相似的类别，梯度下降最终肯定会将它们推得更近，同时它会倾向于将它们从"INLAND"类别的嵌入中移开（见图 13-4）。事实上，表示越好，神经网络就越容易做出准确的预测，因此训练倾向于使嵌入成为类别的有用表示。这个称为表示学习（我们将在第 17 章看到其他类型的表示学习）。

让我们看看我们如何手动实现嵌入，以了解它们是如何工作的（然后我们将使用一个简单的 Keras 层代替）。首先，我们需要创建一个包含每个类别嵌入的嵌入矩阵，随机初始化；每个类别和每个 oov 存储桶将有一行，每个嵌入维度有一列：

embedding_dim = 2
embed_init = tf.random.uniform([len(vocab) + num_oov_buckets, embedding_dim])
embedding_matrix = tf.Variable(embed_init)

在此示例中，我们使用 2D 嵌入，但根据经验，嵌入通常具有 10 到 300 个维度，具体取决于任务和词汇量（您必须调整此超参数）。

这个嵌入矩阵是一个随机的 6 × 2 矩阵，存储在一个变量中（因此可以在训练期间通过梯度下降对其进行调整）：

>>> embedding_matrix
<tf.Variable 'Variable:0' shape=(6, 2) dtype=float32, numpy=
array([[0.6645621 , 0.44100678],
       [0.3528825 , 0.46448255],
       [0.03366041, 0.68467236],
       [0.74011743, 0.8724445 ],
       [0.22632635, 0.22319686],
       [0.3103881 , 0.7223358 ]], dtype=float32)>

现在让我们对与之前相同的一批分类特征进行编码，但这次使用这些嵌入：

>>> categories = tf.constant(["NEAR BAY", "DESERT", "INLAND", "INLAND"])
>>> cat_indices = table.lookup(categories)
>>> cat_indices
<tf.Tensor: id=741, shape=(4,), dtype=int64, numpy=array([3, 5, 1, 1])>
>>> tf.nn.embedding_lookup(embedding_matrix, cat_indices)
<tf.Tensor: id=864, shape=(4, 2), dtype=float32, numpy=
array([[0.74011743, 0.8724445 ],
       [0.3103881 , 0.7223358 ],
       [0.3528825 , 0.46448255],
       [0.3528825 , 0.46448255]], dtype=float32)>

该tf.nn.embedding_lookup()函数在给定索引处查找嵌入矩阵中的行——这就是它所做的一切。例如，查找表表明"INLAND"类别位于索引 1，因此该tf.nn.embedding_lookup()函数返回嵌入矩阵中第 1 行的嵌入（两次）[0.3528825, 0.46448255]：

Keras 提供了一个keras.layers.Embedding层来处理嵌入矩阵（默认情况下可训练）；创建层时，它会随机初始化嵌入矩阵，然后当使用某些类别索引调用它时，它会返回嵌入矩阵中这些索引处的行：

>>> embedding = keras.layers.Embedding(input_dim=len(vocab) + num_oov_buckets,
...                                    output_dim=embedding_dim)
...
>>> embedding(cat_indices)
<tf.Tensor: id=814, shape=(4, 2), dtype=float32, numpy=
array([[ 0.02401174,  0.03724445],
       [-0.01896119,  0.02223358],
       [-0.01471175, -0.00355174],
       [-0.01471175, -0.00355174]], dtype=float32)>

将所有内容放在一起，我们现在可以创建一个 Keras 模型，该模型可以处理分类特征（以及常规数值特征）并学习每个类别（以及每个 oov 桶）的嵌入：

regular_inputs = keras.layers.Input(shape=[8])
categories = keras.layers.Input(shape=[], dtype=tf.string)
cat_indices = keras.layers.Lambda(lambda cats: table.lookup(cats))(categories)
cat_embed = keras.layers.Embedding(input_dim=6, output_dim=2)(cat_indices)
encoded_inputs = keras.layers.concatenate([regular_inputs, cat_embed])
outputs = keras.layers.Dense(1)(encoded_inputs)
model = keras.models.Model(inputs=[regular_inputs, categories],
                           outputs=[outputs])

该模型有两个输入：一个常规输入，每个实例包含八个数字特征，加上一个分类输入（每个实例包含一个分类特征）。它使用一个Lambda层来查找每个类别的索引，然后查找这些索引的嵌入。接下来，它将嵌入和常规输入连接起来，以提供编码输入，这些输入准备好馈送到神经网络。此时我们可以添加任何类型的神经网络，但我们只需添加一个密集输出层，然后创建 Keras 模型。

当keras.layers.TextVectorization图层可用时，您可以调用其adapt()方法以使其从数据样本中提取词汇表（它将负责为您创建查找表）。然后你可以将它添加到你的模型中，它会执行索引查找（替换Lambda前面代码示例中的层）。

现在让我们更仔细地看一下 Keras 预处理层。

Keras 预处理层

这TensorFlow 团队正致力于提供一组标准的Keras 预处理层。当您阅读本文时，它们可能已经可用；但是，届时 API 可能会略有变化，因此如果有任何异常行为，请参阅本章的笔记本。这个新 API 可能会取代现有的 Feature Columns API，后者更难使用且不太直观（如果你想了解更多关于 Feature Columns API 的信息，请查看本章的笔记本）。

我们已经讨论了其中的两个层：keras.layers.Normalization将执行特征标准化的层（它将等同于Standardization我们之前定义的层），以及TextVectorization能够将输入中的每个单词编码到词汇表索引中的层。在这两种情况下，您都会创建层，adapt()使用数据样本调用其方法，然后在模型中正常使用层。其他预处理层将遵循相同的模式。

该 API 还将包括一个keras.layers.Discretization层，该层将连续数据切分到不同的 bin 中，并将每个 bin 编码为 one-hot 向量。例如，您可以使用它将价格离散化为三个类别（低、中、高），这些类别将被编码为 [1, 0, 0]、[0, 1, 0] 和 [0, 0, 1 ]， 分别。当然，这会丢失很多信息，但在某些情况下，它可以帮助模型检测在仅查看连续值时并不明显的模式。

也可以使用PreprocessingStage该类链接多个预处理层。例如，以下代码将创建一个预处理管道，该管道首先对输入进行规范化，然后将它们离散化（这可能会让您想起 Scikit-Learn 管道）。在将此管道调整为数据样本后，您可以像在模型中使用常规层一样使用它（但同样，仅在模型开始时，因为它包含不可微的预处理层）：

normalization = keras.layers.Normalization()
discretization = keras.layers.Discretization([...])
pipeline = keras.layers.PreprocessingStage([normalization, discretization])
pipeline.adapt(data_sample)

该TextVectorization层还可以选择输出字数向量而不是字索引。例如，如果词汇表包含三个单词 say ["and", "basketball", "more"]，那么文本"more and more"将映射到向量[1, 0, 2]：单词"and"出现一次，单词"basketball"根本不出现，单词"more"出现两次。这个文本表示被称为词袋，因为它完全失去了词的顺序。常见的词 like"and"在大多数文本中具有很大的价值，即使它们通常是最不有趣的（例如，在文本"more and more basketball"中，这个词"basketball"显然是最重要的，正是因为它不是一个非常频繁的词）。因此，单词计数应该以一种降低频繁单词重要性的方式进行归一化。这通常使用一种称为 词频×逆文档频率（TF-IDF）。有许多变体，但一个常见的变体是计算出现单词的训练实例的比率，并将单词计数乘以该比率的倒数的对数。例如，假设单词"and"、"basketball"和"more"分别出现在训练集中所有文本实例的 90%、10% 和 50% 中：在这种情况下，最终向量将是[1*log(1/0.9), 0*log(1/0.1), 2*log(1/0.5)]，大约等于[0.1, 0.0, 1.4]。该TextVectorization层将具有执行 TF-IDF 的选项。

如您所见，这些 Keras 预处理层将使预处理变得更加容易！现在，无论您选择编写自己的预处理层还是使用 Keras（甚至使用 Feature Columns API），所有的预处理都将即时完成。然而，在训练期间，最好提前执行预处理。让我们看看我们为什么要这样做以及我们将如何去做。