在Spark程序中使用深度学习模型来处理非结构化数据

随着大数据和AI业务的不断融合，大数据分析和处理过程中，通过深度学习技术多非结构化数据（如图片、音频、文本）的进行大数据处理的业务场景越来越多。本文会介绍Spark如何与深度学习框架进行协同工作，在大数据的处理过程利用深度学习框架对非结构化数据进行处理。

Spark介绍

Spark是大规模数据处理的事实标准，包括机器学习的操作，希望把大数据处理和机器学习管道整合。

Spark使用函数式编程范式扩展了MapReduce模型以支持更多计算类型，可以涵盖广泛的工作流。Spark使用内存缓存来提升性能，因此进行交互式分析也足够快速(就如同使用Python解释器，与集群进行交互一样)。缓存同时提升了迭代算法的性能，这使得Spark非常适合机器学习。

由于Spark库提供了Python、Scale、Java编写的API，以及内建的机器学习、流数据、图算法、类SQL查询等模块；Spark迅速成为当今最重要的分布式计算框架之一。与YARN结合，Spark提供了增量，而不是替代已存在的Hadoop集群。在最近的Spark版本中，Spark加入了对于K8s的支持，为Spark与AI能力的融合提供了更好的支持。

深度学习框架介绍

TensorFlow

TensorFlow 最初是由 Google 机器智能研究部门的 Google Brain 团队开发，基于Google 2011年开发的深度学习基础架构DistBelief构建起来的。由于Google在深度学习领域的巨大影响力和强大的推广能力，TensorFlow一经推出就获得了极大的关注，并迅速成为如今用户最多的深度学习框架。

TensorFlow是一个非常基础的系统，因此也可以应用于众多领域。但由于过于复杂的系统设计，对读者来说，学习TensorFlow底层运行机制更是一个极其痛苦的过程。TensorFlow的接口一直处于快速迭代之中，并且没有很好地考虑向后兼容性，这导致现在许多开源代码已经无法在新版的TensorFlow上运行，同时也间接导致了许多基于TensorFlow的第三方框架出现BUG。

Keras

Keras 于2015年3月首次发布，拥有“为人类而不是机器设计的API”，得到Google的支持。它是一个用于快速构建深度学习原型的高层神经网络库，由纯Python编写而成，以TensorFlow，CNTK，Theano和MXNet为底层引擎，提供简单易用的API接口，能够极大地减少一般应用下用户的工作量。

严格意义上讲，Keras并不能称为一个深度学习框架，它更像一个深度学习接口，它构建于第三方框架之上。Keras的缺点很明显：过度封装导致丧失灵活性。Keras最初作为Theano的高级API而诞生，后来增加了TensorFlow和CNTK作为后端。学习Keras十分容易，但是很快就会遇到瓶颈，因为它缺少灵活性。另外，在使用Keras的大多数时间里，用户主要是在调用接口，很难真正学习到深度学习的内容。

PyTorch

PyTorch于2016年10月发布，是一款专注于直接处理数组表达式的低级API。 前身是 Torch（一个基于 Lua 语言的深度学习库）。Facebook 人工智能研究院对PyTorch提供了强力支持。 PyTorch 支持动态计算图，为更具数学倾向的用户提供了更低层次的方法和更多的灵活性，目前许多新发表的论文都采用PyTorch作为论文实现的工具，成为学术研究的首选解决方案。

Caffe/Caffe2.0

Caffe的全称是Convolutional Architecture for Fast Feature Embedding，它是一个清晰、高效的深度学习框架，于2013年底由加州大学伯克利分校开发，核心语言是C++。它支持命令行、Python和MATLAB接口。Caffe的一个重要特色是可以在不编写代码的情况下训练和部署模型。如果您是C++熟练使用者，并对CUDA计算游刃有余，你可以考虑选择Caffe

在Spark大数据处理中使用深度学习框架

在Spark程序中使用一个预训练过的模型，将其并行应用于大型数据集的数据处理。比如，给定一个可以识别图片的分类模型，其通过一个标准数据集训练（如ImageNet）过。可以在一个Spark程序中调用一个框架（如TensorFlow或keras）进行分布式预测。通过在大数据处理过程中调用预训练模型可以直接对非结构化数据进行直接处理。

我们重点介绍在Spark程序中使用Keras+TensorFlow来进行模型推理。

使用深度学习处理图片的第一步，就是载入图片。Spark 2.3中新增的ImageSchema包含了载入数百万张图像到 Spark DataFrame 的实用函数，并且以分布式方式自动解码，容许可扩展地操做。

使用Spark's ImageSchema：

from pyspark.ml.image import ImageSchema
image_df = ImageSchema.readImages("/data/myimages")
image_df.show()

也可以利用Keras的图片处理库：

from keras.preprocessing import image
img = image.load_img("/data/myimages/daisy.jpg", target_size=(299, 299))

可以通过图片路径来构造Spark DataFrame：

def get_image_paths_df(sqlContext, dirpath, colName):
    files = [os.path.abspath(os.path.join(dirpath, f)) for f in os.listdir(dirpath) if f.endswith('.jpg')]
    return sqlContext.createDataFrame(files, StringType()).toDF(colName)

使用Keras接口加载预训练模型：

from keras.applications import InceptionV3
model = InceptionV3(weights="imagenet")
model.save('/tmp/model-full.h5')
model = load_model('/tmp/model-full.h5')

 
定义图片识别推理方法：

        def iv3_predict(fpath):
            model = load_model('/tmp/model-full.h5')
            img = image.load_img(fpath, target_size=(299, 299))
            x = image.img_to_array(img)
            x = np.expand_dims(x, axis=0)
            x = preprocess_input(x)
 
            preds = model.predict(x)
            preds_decode_list = decode_predictions(preds, top=3)
            tmp = preds_decode_list[0]
            res_list = []
            for x in tmp:
                res = [x[0], x[1], float(x[2])]
                res_list.append(res)
            return res_list

定义推理输入结果Schema

def get_labels_type():    
    ele_type = StructType()    
    ele_type.add("class", data_type=StringType())    
    ele_type.add("description", data_type=StringType())    
    ele_type.add("probability", data_type=FloatType())    
    return ArrayType(ele_type)

将推理方法定义成Spark UDF

spark.udf.register("iv3_predict", iv3_predict, returnType=get_labels_type())

 
载入图片定义为数据表

df = get_image_paths_df(self.sql)
df.createOrReplaceTempView("_test_image_paths_df")

 
使用SQL语句对接图片进行处理

df_images = spark.sql("select fpath, iv3_predict(fpath) as predicted_labels from _test_image_paths_df")
df_images.printSchema()
df_images.show(truncate=False)

结语

在大数据Spark引擎中使用深度学习框架加载预处理模型，来进行非结构数据处理有非常多的应用场景。但是由于深度学习框架的目前比较多，模型与框架本身是深度耦合，在大数据环境中安装和部署深度学习框架软件及其依赖软件会非常复杂，同时不利于大数据集群的管理和维护，增加人力成本。
华为云DLI服务，采用大数据Serverless架构，用户不需要感知实际物理集群，同时DLI服务已经在大数据集群中内置了AI计算框架和底层依赖库（Keras/tensorflow/scikit-learn/pandas/numpy等）。DLI最新版本中已经支持k8s+Docker生态，并开放用户自定义Docker镜像能力，提供给用户来扩展自己的AI框架、模型、算法包。在Serverless基础上，为用户提供更加开放的自定义扩展能力。