第30章 多并行与多线程

Alink任务支持多并行，可以将任务分布到多台机器上，使用更多的CPU和内存资源；每个并发节点还可以通过设置多线程，充分利用CPU的多核资源。

本章主要演示推理计算中如何使用多并行与多线程提高运行速度，使用第13章的KNN模型进行推理计算的场景。

使用第13章所用的MNIST数据，涉及到稀疏向量格式的训练集和测试集，测试集会分别作为批式和流式数据源使用，定义如下面代码所示：

static BatchOperator <?> getTrainSet() {
	return new AkSourceBatchOp().setFilePath(DATA_DIR + SPARSE_TRAIN_FILE);
}

static BatchOperator <?> getTestSet() {
	return new AkSourceBatchOp().setFilePath(DATA_DIR + SPARSE_TEST_FILE);
}

static StreamOperator <?> getTestStream() {
	return new AkSourceStreamOp().setFilePath(DATA_DIR + SPARSE_TEST_FILE);
}

在本章的推理示例中，需要用到KNN模型和Pipeline模型，我们先将模型训练出来并保存到文件，后面的实验可以直接载入模型进行推理计算。具体训练及保存模型的代码如下：

getTrainSet()
	.link(
		new KnnTrainBatchOp()
			.setVectorCol(VECTOR_COL_NAME)
			.setLabelCol(LABEL_COL_NAME)
	)
	.link(
		new AkSinkBatchOp()
			.setFilePath(DATA_DIR + KNN_MODEL_FILE)
			.setOverwriteSink(true)
	);
BatchOperator.execute();

new Pipeline()
	.add(
		new KnnClassifier()
			.setK(3)
			.setVectorCol(VECTOR_COL_NAME)
			.setLabelCol(LABEL_COL_NAME)
			.setPredictionCol(PREDICTION_COL_NAME)
	)
	.fit(getTrainSet())
	.save(DATA_DIR + PIPELINE_MODEL_FILE, true);
BatchOperator.execute();

进一步，定义两个方法，分别获取KNN模型和Pipeline模型，具体代码如下：

static BatchOperator <?> getKnnModel() {
	return new AkSourceBatchOp().setFilePath(DATA_DIR + KNN_MODEL_FILE);
}

static PipelineModel getPipelineModel() {
	return PipelineModel.load(DATA_DIR + PIPELINE_MODEL_FILE);
}