在LifeOmic，机器学习团队经常处理需要复杂特征工程和建模的大型基因组和患者数据集。由于这些数据集的尺寸很大，通过深度学习提取潜在变量(或推断特征)以减少尺寸大小以进行进一步建模通常是很重要的。这包括支持传统技术，如监督学习(预测给定数据的标签)和非监督学习(组织或解释数据)。LifeOmic机器学习的目标是使研究人员能够找到基因组数据和患者数据结合的隐藏见解，而这些数据是无法用基本的统计技术找到的。

拥有有效训练深度学习模式的策略可能是一个挑战。Apache Spark是用于大规模数据处理的统一分析引擎，SparkML是基于Apache Spark构建的机器学习库，它提供了在大型数据集中编排复杂管道的体系结构。然而，Apache Spark目前并没有对深度神经网络体系结构的开箱即用支持，例如用于图像的卷积神经网络、用于自然语言处理的递归神经网络等等。

特征工程管道是训练机器学习模型中经常被忽略的一部分。管道大大简化了将原始数据清理、转换并准备到机器学习模型以执行预测的过程。在LifeOmic里，有了正确的工具，我们可以在原始数据上自动化数据处理步骤，然后把数据发送到模型进行训练或预测。TensorFlow是用于深度学习的高性能数值计算库，对于训练各种神经网络架构非常有用，但与Apache Spark相比，它缺乏对大数据集上管道的特性工程支持。

尽管存在其他开源库来在Apache Spark上训练TensorFlow模型，但很少有人利用SparkML最大的机器学习优势，即将深度学习模型与管道集成起来。

SparkFlow入门

用于数字分类的着名的简单MNIST数据集

SparkFlow利用Spark的管道API中的便捷界面，并将其与TensorFlow结合使用。它可以从Github下载或通过pip安装，使用“pip install sparkflow”。使用SparkFlow在MNIST（用于数字分类）数据集上设置管道的简单Python示例如下所示：

from sparkflow.graph_utils import build_graph

from sparkflow.tensorflow_async import SparkAsyncDL

import tensorflow as tf

from pyspark.ml.feature import VectorAssembler, OneHotEncoder

from pyspark.ml.pipeline import Pipeline

#simple tensorflow network

def small_model():

x = tf.placeholder(tf.float32, shape=[None, 784], name='x')

y = tf.placeholder(tf.float32, shape=[None, 10], name='y')

layer1 = tf.layers.dense(x, 256, activation=tf.nn.relu)

layer2 = tf.layers.dense(layer1, 256, activation=tf.nn.relu)

out = tf.layers.dense(layer2, 10)

z = tf.argmax(out, 1, name='out')

loss = tf.losses.softmax_cross_entropy(y, out)

return loss

df = spark.read.option("inferSchema", "true").csv('mnist_train.csv')

#convert graph to json

tensorflow_graph = build_graph(small_model)

#Assemble and one hot encode

va = VectorAssembler(inputCols=df.columns[1:785], outputCol='features')

encoded = OneHotEncoder(inputCol='_c0', outputCol='labels', dropLast=False)

spark_model = SparkAsyncDL(

inputCol='features',

tensorflowGraph=tensorflow_graph,

tfInput='x:0',

tfLabel='y:0',

tfOutput='out:0',

tfLearningRate=.001,

iters=20

)

p = Pipeline(stages=[va, encoded, spark_model]).fit(df)

说明

为了完全理解示例中发生了什么，有必要浏览一下SparkFlow部分

def small_model():

x = tf.placeholder(tf.float32, shape=[None, 784], name='x')

y = tf.placeholder(tf.float32, shape=[None, 10], name='y')

layer1 = tf.layers.dense(x, 256, activation=tf.nn.relu)

layer2 = tf.layers.dense(layer1, 256, activation=tf.nn.relu)

out = tf.layers.dense(layer2, 10)

z = tf.argmax(out, 1, name='out')

loss = tf.losses.softmax_cross_entropy(y, out)

return loss

下面突出显示的small_model函数封装了TensorFlow graph。该图是一个简单的，全连接的网络，每个隐藏层有256个神经元。“out”变量表示将用于损失函数的logits或raw输出，变量“z”表示将用于Spark Transformer中的预测的argmax。最后，该函数返回损失变量，该变量将由指定的优化器最小化。

tensorflow_graph = build_graph(small_model)

一旦small_model函数完成，它就可以传递给效用函数“build_graph”，它将TensorFlow图序列化。SparkFlow需要一个序列化图，以便它可以将网络广播到Spark执行程序

spark_model = SparkAsyncDL(

inputCol='features',

tensorflowGraph=tensorflow_graph,

tfInput='x:0',

tfLabel='y:0',

tfOutput='out:0',

tfLearningRate=.001,

iters=20)

最后，我们可以构建我们的模型。还有几个可以传递给SparkAsyncDL类的选项，包括不同的优化器（目前支持从版本1.8.0开始的所有TensorFlow优化器），批处理选项等。可以在此处找到这些选项的完整列表(https://github.com/lifeomic/sparkflow#using-sparkasyncdl-and-options)。

SparkFlow架构

SparkFlow利用Spark的驱动程序/执行程序体系结构，将驱动程序用作参数服务器，将执行程序用作工作程序。当然，这创建了一个实现异步训练算法的简单途径，例如Hogwild。Hogwild算法通过在主节点上存在神经网络以及在每个任务实例上存储该网络的副本来并行化神经网络的训练。然后，任务实例计算一批数据的渐变，并将它们发送给驱动程序以更新权重。Hogwild与异步训练不同，因为它规定了更新主网络梯度时的无锁方法。（作为旁注，SparkFlow支持使用锁和无锁方法进行异步训练）。

以SparkFlow的Hogwild实现为例，每个Spark分区都获取TensorFlow图和权重的副本，然后计算一小批数据的损失和梯度，并将信息发送给驱动程序以更新主网络。在用户指定的迭代次数完成后，工作人员从参数服务器中提取权重以更新其网络副本。然后它可以恢复训练。

具有在大型数据集上跨多个CPU或GPU训练网络的能力，可以加快训练和收敛速度。使用SparkFlow训练模型后，可以将其保存为常规管道并用于预测。用户可以将他们的特征工程工作与他们的TensorFlow模型组合在一个完整的包中。

从版本0.3开始，SparkFlow现在支持引入预先训练的TensorFlow模型并将它们附加到基于Spark的管道。因此，现有模型可用于管道内的预测，这对于具有现有Spark管道的公司而言可能是有价值的。

SparkFlow走向

SparkFlow的未来计划包括针对大型数据集的超参数优化实现、第二个api紧密反映Spark现有的estimator api，以及对新的TensorFlow概率api的一级支持。文档和更多教程也将继续是优先级。

最后

随着更大的数据集变得可用，能够在Apache Spark等分布式计算系统上运行深度学习模型非常重要。SparkFlow使用户不仅可以在Spark中训练深度学习模型，还可以将训练好的模型附加到管道上，以便对原始数据进行无缝预测。