在本章的第2、3节介绍了使用Alink提供的深度学习组件KerasSequentialClassifier和KerasSequentialRegressor进行分类和回归模型的训练、预测。
实际应用中，经常需要使用TensorFlow或着PyTorch训练好的模型，对流式数据、批式数据进行预测。Alink提供了相应的流式、批式和Pipeline组件适配TensorFlow或着PyTorch模型。
本节重点介绍与PyTorch模型相关的操作。

25.5.1 生成PyTorch模型

本节所需的PyTorch模型文件mnist_model_pytorch.pt，已经被放到了OSS上，本节后面的实验会直接从网络读取该模型。https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt

如果读者有兴趣，可以在PyTorch环境，运行下面代码便可生成PyTorch模型，从而被Alink相关组件使用。注意：PyTorch模型执需要打包为”.pt”文件，便于Alink相关组件导入模型。建议的打包示例代码在下面代码的最后部分。

  1. import torch
  2. from torchvision import datasets
  3. from torchvision.transforms import ToTensor
  5. train_data = datasets.MNIST(
  6.     root="data",
  7.     train=True,
  8.     download=True,
  9.     transform=ToTensor()
  10. )
  12. train_loader = torch.utils.data.dataloader.DataLoader(dataset=train_data, batch_size=64, shuffle=True)
  14. class Net(torch.nn.Module):
  15.     def __init__(self):
  16.         super(Net, self).__init__()
  17.         self.conv1 = torch.nn.Sequential(
  18.             torch.nn.Conv2d(1, 32, 3, 1, 1),
  19.             torch.nn.ReLU(),
  20.             torch.nn.MaxPool2d(2))
  21.         self.conv2 = torch.nn.Sequential(
  22.             torch.nn.Conv2d(32, 64, 3, 1, 1),
  23.             torch.nn.ReLU(),
  24.             torch.nn.MaxPool2d(2)
  25.         )
  26.         self.conv3 = torch.nn.Sequential(
  27.             torch.nn.Conv2d(64, 64, 3, 1, 1),
  28.             torch.nn.ReLU(),
  29.             torch.nn.MaxPool2d(2)
  30.         )
  31.         self.dense = torch.nn.Sequential(
  32.             torch.nn.Linear(64 * 3 * 3, 128),
  33.             torch.nn.ReLU(),
  34.             torch.nn.Linear(128, 10)
  35.         )
  37.     def forward(self, x):
  38.         conv1_out = self.conv1(x)
  39.         conv2_out = self.conv2(conv1_out)
  40.         conv3_out = self.conv3(conv2_out)
  41.         res = conv3_out.view(conv3_out.size(0), -1)
  42.         out = self.dense(res)
  43.         return out
  45. model = Net()
  46. print(model)
  48. optimizer = torch.optim.Adam(model.parameters())
  49. loss_func = torch.nn.CrossEntropyLoss()
  51. for epoch in range(5):
  52.     print('epoch {}'.format(epoch + 1))
  53.     train_loss = 0.
  54.     train_acc = 0.
  55.     for batch_x, batch_y in train_loader:
  56.         batch_x, batch_y = torch.autograd.Variable(batch_x), torch.autograd.Variable(batch_y)
  57.         out = model(batch_x)
  58.         loss = loss_func(out, batch_y)
  59.         train_loss += loss.item()
  60.         pred = torch.max(out, 1)[1]
  61.         train_correct = (pred == batch_y).sum()
  62.         train_acc += train_correct.item()
  63.         optimizer.zero_grad()
  64.         loss.backward()
  65.         optimizer.step()
  66.     print('Train Loss: {:.6f}, Acc: {:.6f}'.format(train_loss / (len(
  67.         train_data)), train_acc / (len(train_data))))
  70. traced = torch.jit.trace(model, (torch.rand(1, 1, 28, 28)))
  71. torch.jit.save(traced, "mnist_model_pytorch.pt")

输出模型及训练信息如下：

  1. Net(
  2.   (conv1): Sequential(
  3.     (0): Conv2d(1, 32, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
  4.     (1): ReLU()
  5.     (2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  6.   )
  7.   (conv2): Sequential(
  8.     (0): Conv2d(32, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
  9.     (1): ReLU()
  10.     (2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  11.   )
  12.   (conv3): Sequential(
  13.     (0): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
  14.     (1): ReLU()
  15.     (2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  16.   )
  17.   (dense): Sequential(
  18.     (0): Linear(in_features=576, out_features=128, bias=True)
  19.     (1): ReLU()
  20.     (2): Linear(in_features=128, out_features=10, bias=True)
  21.   )
  22. )
  23. epoch 1
  24. Train Loss: 0.003327, Acc: 0.933000
  25. epoch 2
  26. Train Loss: 0.000876, Acc: 0.982667
  27. epoch 3
  28. Train Loss: 0.000592, Acc: 0.987783
  29. epoch 4
  30. Train Loss: 0.000466, Acc: 0.990467
  31. epoch 5
  32. Train Loss: 0.000403, Acc: 0.991733

25.5.2 批式任务中使用PyTorch模型

使用TorchModelPredictBatchOp组件，可以加载PyTorch模型进行批式预测。关于该组件的详细说明参见Alink文档 https://www.yuque.com/pinshu/alink_doc/torchmodelpredictbatchop .
使用PyTorch模型前，还需要将输入数据列的类型转换为Tensor格式，可以使用VectorToTensorBatchOp组件。具体代码如下所示：

  1. AkSourceBatchOp()\
  2.     .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\
  3.     .link(\
  4.         VectorToTensorBatchOp()\
  5.             .setTensorDataType("float")\
  6.             .setTensorShape([1, 1, 28, 28])\
  7.             .setSelectedCol("vec")\
  8.             .setOutputCol("tensor")\
  9.             .setReservedCols(["label"])
  10.     )\
  11.     .link(\
  12.         TorchModelPredictBatchOp()\
  13.             .setModelPath(
  14.                 "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt")\
  15.             .setSelectedCols(["tensor"])\
  16.             .setOutputSchemaStr("output_1 FLOAT_TENSOR")
  17.     )\
  18.     .lazyPrint(3)\
  19.     .link(\
  20.         UDFBatchOp()\
  21.             .setFunc(get_max_index)\
  22.             .setSelectedCols(["output_1"])\
  23.             .setOutputCol("pred")
  24.     )\
  25.     .lazyPrint(3)\
  26.     .link(\
  27.         EvalMultiClassBatchOp()\
  28.             .setLabelCol("label")\
  29.             .setPredictionCol("pred")\
  30.             .lazyPrintMetrics()
  31.     )
  33. BatchOperator.execute()

这里用到了一个自定义函数，具体定义如下：

  1. import numpy as np
  3. @udf(input_types=[AlinkDataTypes.TENSOR()], result_type=AlinkDataTypes.INT()) 
  4. def get_max_index(tensor: np.ndarray):
  5.     return tensor.argmax().item()

批式任务的运行结果为：
image.png

  1. -------------------------------- Metrics: --------------------------------
  2. Accuracy:0.9903    Macro F1:0.9902    Micro F1:0.9903    Kappa:0.9892    
  3. |Pred\Real|  9|  8|   7|...|   2|   1|  0|
  4. |---------|---|---|----|---|----|----|---|
  5. |        9|992|  1|   4|...|   0|   0|  0|
  6. |        8|  2|965|   1|...|   0|   1|  1|
  7. |        7|  5|  2|1012|...|   2|   0|  1|
  8. |      ...|...|...| ...|...| ...| ...|...|
  9. |        2|  2|  4|   9|...|1030|   3|  2|
  10. |        1|  0|  0|   2|...|   0|1128|  0|
  11. |        0|  0|  2|   0|...|   0|   0|973|

25.5.3 流式任务中使用PyTorch模型

使用TorchModelPredictStreamOp组件，可以加载PyTorch模型进行批式预测。关于该组件的详细说明参见Alink文档 https://www.yuque.com/pinshu/alink_doc/torchmodelpredictstreamop .
使用PyTorch模型前，还需要将输入数据列的类型转换为Tensor格式，可以使用VectorToTensorStreamOp组件。具体代码如下所示：

  1. AkSourceStreamOp()\
  2.     .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\
  3.     .link(\
  4.         VectorToTensorStreamOp()\
  5.             .setTensorDataType("float")\
  6.             .setTensorShape([1, 1, 28, 28])
  7.             .setSelectedCol("vec")\
  8.             .setOutputCol("tensor")\
  9.             .setReservedCols(["label"])
  10.     )\
  11.     .link(\
  12.         TorchModelPredictStreamOp()\
  13.             .setModelPath(
  14.                 "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt")\
  15.             .setSelectedCols(["tensor"])\
  16.             .setOutputSchemaStr("output_1 FLOAT_TENSOR")
  17.     )\
  18.     .link(\
  19.         UDFStreamOp()\
  20.             .setFunc(get_max_index)\
  21.             .setSelectedCols(["output_1"])\
  22.             .setOutputCol("pred")
  23.     )\
  24.     .sample(0.001)\
  25.     .print()
  27. StreamOperator.execute()

运行结果为：
image.png

25.5.4 Pipeline中使用PyTorch模型

学习了如何在批式任务和流式任务中使用PyTorch模型，我们很容易在Pipeline中使用PyTorch模型进行预测，只要将其中的批式/流式组件对应到Pipeline组件即可。具体代码如下：

  1. PipelineModel(\
  2.     VectorToTensor()\
  3.         .setTensorDataType("float")\
  4.         .setTensorShape([1, 1, 28, 28])\
  5.         .setSelectedCol("vec")\
  6.         .setOutputCol("tensor")\
  7.         .setReservedCols(["label"]),
  8.     TorchModelPredictor()\
  9.         .setModelPath(
  10.             "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt")\
  11.         .setSelectedCols(["tensor"])\
  12.         .setOutputSchemaStr("output_1 FLOAT_TENSOR")
  13. ).save(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL, True)
  14. BatchOperator.execute()
  16. PipelineModel\
  17.     .load(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL)\
  18.     .transform(\
  19.         AkSourceStreamOp()\
  20.             .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)
  21.     )\
  22.     .link(\
  23.         UDFStreamOp()\
  24.             .setFunc(get_max_index)\
  25.             .setSelectedCols(["output_1"])\
  26.             .setOutputCol("pred")
  27.     )\
  28.     .sample(0.001)\
  29.     .print()
  30. StreamOperator.execute()

运行结果为：
image.png

25.5.5 LocalPredictor中使用PyTorch模型

除了通过Alink任务使用PyTorch模型，也可以使用LocalPredictor进行嵌入式预测。示例代码如下，首先从数据集中抽取一行数据，输入数据的SchemaStr为“vec string, label int”；然后通过导入上一节保存的Pipeline模型，并设置输入数据的SchemaStr，得到LocalPredictor类型的实例localPredictor；如果不确定预测结果各列的含义，可以打印输出localPredictor的OutputSchema；使用localPredictor的map方法获得预测结果。

  1. source = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)
  3. print(source.getSchemaStr())
  5. df = source.firstN(1).collectToDataframe()
  7. row = [df.iat[0,0], df.iat[0,1].item()]
  9. localPredictor = LocalPredictor(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL, "vec string, label int")
  11. print(localPredictor.getOutputSchemaStr())
  13. r = localPredictor.map(row)
  14. print(str(r[0]) + " | " + str(r[2]))

运行结果为：

  1. vec VARCHAR, label INT
  2. label INT, tensor ANY<com.alibaba.alink.common.linalg.tensor.FloatTensor>, output_1 ANY<com.alibaba.alink.common.linalg.tensor.FloatTensor>
  3. 2 | FloatTensor(1,10)
  4. [[500.5041 499.77554 4562.968 ... -759.00934 -1826.2468 -2071.0444]]