Tensorflow多输入模型构建以及Dataset数据构建

多输入模型多适用于问答模型或者对于时间序列模型来说有部分特征是针对样本个体而固定的，不随时间变换而发生改变的情况下。
对于模型的输入数据格式来说，有很多种方式，普通的全部数据导入，或者写成生成器等，可以逐批读取数据然后训练模型，但是当你使用tensorflow内置分布式训练，也就是多机多卡模卡MultiWorkerMirroredStrategy的时候，就必须使用Dataset格式。
因为Dataset会自动根据batch_size分发数据进行迭代训练。

如果对MultiWorkerMirroredStrategy以及MirroredStrateg两种训练模型感兴趣的可以去看看我的另外一篇文章。https://blog.csdn.net/qq_35869630/article/details/106313745

一、多输入模型

import tensorflow as tf
from tensorflow.keras.layers import LSTM, Dense, concatenate,Bidirectional
from tensorflow.keras import Input,Modeldef build_and_compile_model():inputA = Input(shape=(7, 22))inputB = Input(shape=(3,))x = LSTM(128, return_sequences=False, activation="relu")(inputA)x = Dense(128, activation="relu")(x)x = Dense(64, activation="relu")(x)x = Model(inputs=inputA, outputs=x)y = Dense(32, activation="relu")(inputB)y = Dense(16, activation="relu")(y)y = Model(inputs=inputB, outputs=y)combined = concatenate([x.output, y.output], axis=-1)z = Dense(32, activation="relu")(combined)z = Dense(1, activation="sigmoid")(z)model = Model(inputs=[x.input, y.input], outputs=z)model.compile(loss='binary_crossentropy', optimizer='rmsprop',metrics=['accuracy'])# rmspropmodel.summary()return model

介绍一下背景，我们这里构建的模型主要是为了用于观察某个玩家7天内的22个特征数据变化，以及这个玩家的3个原本固定属性，来判断这个玩家是否违反了规定。

inputA 表示的是对时间序列长度为7，特征数为22的输入数据。
inputB 表示的是对特征数为3个的数据的输入。
x = Model(inputs=inputA, outputs=x)
y = Model(inputs=inputB, outputs=y)
构建过单输入模型的都懂这是将模型建立起来，设置输入输出层。
concatenate 将两个模型联结起来。axis有多种取值。
axis=n表示从第n个维度进行拼接，对于一个三维矩阵，axis的取值可以为[-3, -2, -1, 0, 1, 2]。虽然keras用模除允许axis的取值可以在这个范围之外，但不建议那么用。
如果看不懂可以看看大佬介绍的
https://blog.csdn.net/leviopku/article/details/82380710

简单的结构如下

这样就完成了多输入模型的构建。
那针对这样的对输入模型，怎么构建相对应的dataset呢？

二、Dataset

def create_dataset(batch_size,num_workers,clean_data):train_x, train_s_x, train_y, test_x, test_s_x, test_y = clean_data[:]# 转成 datasetg_batch_size = batch_size*num_workerssteps_per_epoch_size = int(train_x.shape[0] / g_batch_size)test_steps = int(test_x.shape[0] / g_batch_size)def func_s(x1, x2):return [x1, x2]input_dataset = tf.data.Dataset.zip((tf.data.Dataset.from_tensor_slices((train_x, train_s_x)).map(func_s),tf.data.Dataset.from_tensor_slices(train_y)))train_datasets_unbatched = input_dataset.cache().shuffle(1000)train_datasets = train_datasets_unbatched.batch(g_batch_size).repeat(-1)test_input_dataset = tf.data.Dataset.zip((tf.data.Dataset.from_tensor_slices((test_x, test_s_x)).map(func_s),tf.data.Dataset.from_tensor_slices(test_y)))test_input_dataset = test_input_dataset.cache().shuffle(1000)test_datasets = test_input_dataset.batch(g_batch_size).repeat(-1)return train_datasets,test_datasets,steps_per_epoch_size,test_steps

因为是多输入模型，所以这里用到了 tf.data.Dataset.zip 将两个输入合并起来。
repeat（-1）设置为无限循环取数。

这里其他函数也挺简单的就不多做介绍了。

直接看全部完整的代码。

def get_multi_strategy():"""分布式模式:return:"""import tensorflow as tfimport argparseimport osimport jsonparser = argparse.ArgumentParser(description='tensorflow_test')parser.add_argument('-worker', default=0)  # --work_numargs, unknown_args = parser.parse_known_args()HOST_CONFIG = ip_listos.environ['TF_CONFIG'] = json.dumps({'cluster': {'worker': HOST_CONFIG},'task': {'type': 'worker', 'index': args.worker}})print (os.environ['TF_CONFIG'])strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()return strategydef build_and_compile_model():inputA = Input(shape=(7, 22))inputB = Input(shape=(3,))x = LSTM(128, return_sequences=False, activation="relu")(inputA)x = Dense(128, activation="relu")(x)x = Dense(64, activation="relu")(x)x = Model(inputs=inputA, outputs=x)y = Dense(32, activation="relu")(inputB)y = Dense(16, activation="relu")(y)y = Model(inputs=inputB, outputs=y)combined = concatenate([x.output, y.output], axis=0)z = Dense(32, activation="relu")(combined)z = Dense(1, activation="sigmoid")(z)model = Model(inputs=[x.input, y.input], outputs=z)model.compile(loss='binary_crossentropy', optimizer='rmsprop',metrics=['accuracy'])# rmspropmodel.summary()return model# batch_size 单机训练的size，num_workers 机器个数
def create_dataset(batch_size,num_workers,clean_data):train_x, train_s_x, train_y, test_x, test_s_x, test_y = clean_data[:]# 转成 datasetg_batch_size = batch_size*num_workerssteps_per_epoch_size = int(train_x.shape[0] / g_batch_size)test_steps = int(test_x.shape[0] / g_batch_size)def func_s(x1, x2):return [x1, x2]input_dataset = tf.data.Dataset.zip((tf.data.Dataset.from_tensor_slices((train_x, train_s_x)).map(func_s),tf.data.Dataset.from_tensor_slices(train_y)))train_datasets_unbatched = input_dataset.cache().shuffle(1000)train_datasets = train_datasets_unbatched.batch(g_batch_size).repeat(-1)test_input_dataset = tf.data.Dataset.zip((tf.data.Dataset.from_tensor_slices((test_x, test_s_x)).map(func_s),tf.data.Dataset.from_tensor_slices(test_y)))test_input_dataset = test_input_dataset.cache().shuffle(1000)test_datasets = test_input_dataset.batch(g_batch_size).repeat(-1)return train_datasets,test_datasets,steps_per_epoch_size,test_stepsif __name__ == '__main__':sequence_length = 7batch_size = 64num_workers = 4epochs = 50strategy = get_multi_strategy()clean_d = clean_data()c_data = train_data(clean_d, sequence_length)train_datasets, test_datasets, steps_per_epoch_size, test_steps = create_dataset(batch_size, num_workers, c_data)print("steps_per_epoch_size,test_steps:", steps_per_epoch_size, test_steps)with strategy.scope():model = build_and_compile_model()print('start fit model')history = model.fit(train_datasets, epochs=epochs, steps_per_epoch=steps_per_epoch_size, shuffle=True,validation_data=test_datasets, validation_steps=test_steps)

三、总结

tensorflow多输入多输出模型结构具备多样化，dataset也要跟着进行改变，如果你不采用MultiWorkerMirroredStrategy多机多卡模型的话也可以不用dataset数据格式，根据实际情况进行调整。

顺便介绍一下MultiWorkerMirroredStrategy多机多卡模式确实能够在机器不行的情况下通过多加机器的方法起到加速模型训练的作用，但是由于还在实验阶段，并不具备所有完整的函数，因此使用的时候也要稍加注意。

我是一只前进的蚂蚁，希望能一起前行。

如果对您有一点帮助，一个赞就够了，感谢！

注：如果本篇博客有任何错误和建议，欢迎各位指出，不胜感激！！！