Spark新愿景：让深度学习变得更加易于使用

   <h2>前言</h2>    <p>Spark成功的实现了当年的承诺，让数据处理变得更容易，现在，雄心勃勃的Databricks公司展开了一个新的愿景：让深度学习变得更容易。 当然牛好吹，也是要做些实际行动的，所有便有了 <a href="/misc/goto?guid=4959754627883633765" rel="nofollow,noindex">spark-deep-learning</a> 项目。这件事情已经有很多人尝试做了，但显然太浅了，DB公司则做的更深入些。</p>    <h2>原理</h2>    <p>要做深度学习，肯定不能离开TensorFlow, MXNet之类的。 spark-deep-learning也是如此，尝试和Tensorflow进行整合。那么如何进行整合呢？ 我们知道Tensorflow其实是C++开发的，平时训练啥的我们主要使用python API。Spark要和TensorFlow 进行整合，那么有三种方式：</p>    <ol>     <li>走Tensorflow的Java API</li>     <li>走Tensorflow的Python API</li>     <li>通过JNI直接走Tensorflow的C++ API</li>    </ol>    <p>因为Spark自己也可以使用Python，虽然有性能的上的损耗（据说>30%）,但是终究是能跑起来。实际上Spark采用了2和3的结合。 第二条容易理解，第三条则主要依赖于另外一个项目 <a href="/misc/goto?guid=4959741470145431012" rel="nofollow,noindex">tensorframes</a> 。这个项目主要是实现tensorflow和spark的互相调用。简单的来说，在spark的dataframe运算可以通过JNI调用tensorflow来完成，反之Spark的dataframe也可以直接喂给tensorflow(也就是tensorflow可以直接输入dataframe了)。有了这个之后，spark-deep-learning 则无需太多关注如何进行两个系统完成交互的功能，而是专注于完成对算法的集成了。</p>    <p>为了给出一个直观的感受，我们看个示例代码（来源于官方）：</p>    <pre>  <code class="language-python">import tensorflow as tf  import tensorframes as tfs  from pyspark.sql import Row    data = [Row(x=float(x)) for x in range(10)]  df = sqlContext.createDataFrame(data)  with tf.Graph().as_default() as g:      # The TensorFlow placeholder that corresponds to column 'x'.      # The shape of the placeholder is automatically inferred from the DataFrame.      x = tfs.block(df, "x")      # The output that adds 3 to x      z = tf.add(x, 3, name='z')      # The resulting dataframe      df2 = tfs.map_blocks(z, df)    # The transform is lazy as for most DataFrame operations. This will trigger it:  df2.collect()</code></pre>    <p>在这里，通过tensorframes 我可以对spark dataframe里列使用tensorflow来进行处理。</p>    <pre>  <code class="language-python">x = tfs.block(df, "x")</code></pre>    <p>相当于</p>    <pre>  <code class="language-python">x =  tf.placeholder(shape=..., dtype=..., name='x')</code></pre>    <p>程序自动从df可以知道数据类型。</p>    <pre>  <code class="language-python">df2 = tfs.map_blocks(z, df)</code></pre>    <p>则相当于将df 作为tf的feed_dict数据。最终f2.collect 触发实际的计算。</p>    <p>spark-deep-learning 提出了三个新的东西：</p>    <ol>     <li>首先是，Spark的数据终于可以用DF的方式无缝的喂给Tensorflow/Keras了，而且对Tensorflow/Keras的适配了一套Mllib的库，方便以Spark Mllib的方式进行编程。当然，为了使得原先是Tensorflow/Keras的用户感觉爽，如果你使用Python API你也可以完全使用Keras/Tensorflow 的Style来完成代码的编写。</li>     <li>其次是多个TF模型同时训练，给的一样的数据，但是不同的参数，从而充分利用分布式并行计算来选择最好的模型。</li>     <li>另外是模型训练好后如何集成到Spark里进行使用呢？没错，SQL UDF函数，你可以很方便的把一个训练好的模型注册成UDF函数，从而实际完成了模型的部署。</li>    </ol>    <p>方便理解，我们也简单看看一些代码：</p>    <pre>  <code class="language-python">from pyspark.ml.classification import LogisticRegression  from pyspark.ml.evaluation import MulticlassClassificationEvaluator  from pyspark.ml import Pipeline  from sparkdl import DeepImageFeaturizer  from sparkdl import readImages  from pyspark.sql.functions import lit        //读取图片，设置为1分类  tulips_df = readImages(img_dir + "/tulips").withColumn("label", lit(1))  //读取图片，设置为2分类  daisy_df = readImages(img_dir + "/daisy").withColumn("label", lit(0))  //构成训练集  train_df = tulips_train.unionAll(daisy_train)    //使用已经配置好的模型(InceptionV3)  featurizer = DeepImageFeaturizer(inputCol="image", outputCol="features", modelName="InceptionV3")    //接一个分类器，也就是传说中的迁移学习  lr = LogisticRegression(maxIter=20, regParam=0.05, elasticNetParam=0.3, labelCol="label")  //组装下  p = Pipeline(stages=[featurizer, lr])  //训练，和Mllib保持了一致  model = p.fit(image_df)    # train_images_df is a dataset of images (SpImage) and labels    //预测  df = model.transform(train_df.limit(10)).select("image", "probability",  "uri", "label")  predictionAndLabels = df.select("prediction", "label")</code></pre>    <p>整个模型一气呵成。</p>    <h2>如何开发</h2>    <p>spark-deep-learning 还处于早期，很多东西还不太完善。</p>    <p>为了方便看源码以及编写实际的代码，你可以clone最新的代码，然后使用intellij idea 可以很方便的导入进来。导入进来后，添加python framework的支持，然后把根目录下的python目录作为source 目录，接着进入project structured 添加pyspark 的zip（一般放在spark home 里的lib目录），这样你在spark-deep-learning里就可以直接做开发了。</p>    <p>spark-deep-learning使用的是spark 2.1.1 以及python 2.7 ，不过我的环境是spark 2.2.0, python 3.6。 所以你需要在build.sbt里第一行修改为</p>    <pre>  <code class="language-python">val sparkVer = sys.props.getOrElse("spark.version", "2.2.0")</code></pre>    <p>同时保证你的python为2.7版本（你可以通过一些python的管理工具来完成版本的切换），然后进行编译：</p>    <pre>  <code class="language-python">build/sbt assembly</code></pre>    <p>编译的过程中会跑单元测试，在spark 2.2.0会报错，原因是udf函数不能包含“-”，所以你找到对应的几个测试用例，修改里面的udf函数名称即可。</p>    <p>编译好后，你就可以直接写个脚本，比如：</p>    <pre>  <code class="language-python">import os  from pyspark import *  from sparkdl import readImages    os.environ['PYSPARK_PYTHON'] = '/Users/allwefantasy/python2.7/tensorflow/bin/python'    sc = SparkContext.getOrCreate()    image_df = readImages("/Users/allwefantasy/resources/images/flower_photos/daisy/")  image_df.show()</code></pre>    <p>比如我这里简单的读取图片文件，并且显示出来。你可以直接点击右键运行，也可以通过spark-submit运行：</p>    <pre>  <code class="language-python">./bin/spark-submit --driver-memory 8g    --py-files spark-deep-learning-assembly-0.1.0-spark2.2.jar  \  --jars spark-deep-learning-assembly-0.1.0-spark2.2.jar  \  --master local[*]  spark-deep-learning/python/tests/Test.py</code></pre>    <p>因为比较消耗内存，这里可以通过driver-memory 设置spark submit 内存。</p>    <p> </p>    <p>来自：http://www.jianshu.com/p/07e8200b7cea</p>    <p> </p>