【導(dǎo)讀】你能利用現(xiàn)有的 Spark 集群構(gòu)建深度學(xué)習(xí)模型嗎？如何分析存儲(chǔ)在 HDFS、Hive 和 HBase 中 tb 級(jí)的數(shù)據(jù)嗎？企業(yè)想用深度學(xué)習(xí)模型，可是要考慮的問(wèn)題又很多，怎么破？這篇文章中，我們將給大家講講大數(shù)據(jù)+深度學(xué)習(xí)下，BigDL 框架的利弊與應(yīng)用教程，為什么有了 TF、PyTorch，還是會(huì)考慮用 BigDL？

為什么要講 BigDL？

這幾年，曾被稱為 “3S”，因其簡(jiǎn)單、快速并支持深度學(xué)習(xí)的 Apache Spark 非常流行。許多公司利用 Hadoop 和 Spark 環(huán)境來(lái)構(gòu)建強(qiáng)大的數(shù)據(jù)處理 pipeline，對(duì)分布式集群上的大量數(shù)據(jù)進(jìn)行預(yù)處理，并從中挖掘出業(yè)務(wù)提升的新觀點(diǎn)。現(xiàn)在許多公司都希望能利用深度學(xué)習(xí)的模型幫助自己進(jìn)一步改善業(yè)務(wù)。雖然深度學(xué)習(xí)模型的性能在不斷提高，但是想要在現(xiàn)有的平臺(tái)上部署新技術(shù)也還有很多問(wèn)題需要權(quán)衡，比如：

（1）如果用深度學(xué)習(xí)的方法，還可以利用原有的 pipeline 嗎？

（2）當(dāng)深度學(xué)習(xí)遇到大規(guī)模數(shù)據(jù)集時(shí)，“大規(guī)模深度學(xué)習(xí)”如何能保證其有效性？

（3）基于現(xiàn)有的 Spark / Hadoop 集群是否可以用？

為什么要權(quán)衡這些問(wèn)題其實(shí)不難理解，我們需要保持一致的環(huán)境，避免大型數(shù)據(jù)集跨不同集群之間的傳遞。此外，從現(xiàn)有的基礎(chǔ)設(shè)施中移動(dòng)專有數(shù)據(jù)集也有安全風(fēng)險(xiǎn)與隱患。早期時(shí)解決這些問(wèn)題的方法是在 Spark 上直接加入深度學(xué)習(xí)框架，但并不能保證保持它們之間的一致性，因此，后來(lái)產(chǎn)生了基于 Spark 的 BigDL 平臺(tái)，其繼承了 3S 的主要特點(diǎn)：簡(jiǎn)單、快速、支持深度學(xué)學(xué)習(xí)。

提到 BigDL 框架，也許大家對(duì)他的熟悉度不高，下面我們就先為大家簡(jiǎn)單的介紹一下什么是 BigDL 框架。

BigDL 是一個(gè)分布式的深度學(xué)習(xí)框架，在大數(shù)據(jù)分析領(lǐng)域發(fā)展迅速，并且也是一個(gè)開(kāi)源的框架。BigDL 有很多特點(diǎn)，比如：與 Spark 和 Hadoop 生態(tài)系統(tǒng)進(jìn)行了完整集成，具有可拓展性等很多重要的功能。可根據(jù)數(shù)據(jù)大小在任意集群中訓(xùn)練模型、支持構(gòu)建端到端的大數(shù)據(jù)分析與深度學(xué)習(xí)等 pipeline、可執(zhí)行數(shù)據(jù)并行分布式訓(xùn)練，實(shí)現(xiàn)高可擴(kuò)展性。BigDL 用戶可在 Spark 和大數(shù)據(jù)平臺(tái)上構(gòu)建了大量數(shù)據(jù)分析與深度學(xué)習(xí)的應(yīng)用，如視覺(jué)相似性、參數(shù)同步、比例縮放等。

深度學(xué)習(xí)應(yīng)用程序可以編寫為標(biāo)準(zhǔn)的 spark 庫(kù)。這些 Spark 框架中統(tǒng)一的庫(kù)可以讀取大量數(shù)據(jù)。此外，它還支持 Numpy、Scipy、NLTK、Pandas 等 Python 庫(kù)；與 TensorBoard 集成用于可視化分析；支持加載現(xiàn)有的 Torch 模型。企業(yè)客戶使用 BigDL 和Spark 還有一個(gè)重要的原因，相比 TensorFlow，BigDL 不僅更快，通過(guò)并行計(jì)算它能夠更快地重新訓(xùn)練模型。

分享一位網(wǎng)友對(duì) BigDL 的總結(jié)：

BigDL相對(duì)于其他主流的深度學(xué)習(xí)框架(TensorFlow/Caffe/PyTorch)，算是一個(gè)異類。其異有二：（1）CPU、（2）純分布式(Spark)

雖然業(yè)界普遍不看好CPU跑深度學(xué)習(xí)，但實(shí)際上還是有需求的。比如，現(xiàn)有Hadoop集群的公司，復(fù)用現(xiàn)有集群來(lái)跑深度學(xué)習(xí)是最經(jīng)濟(jì)的方案。

并且，充分優(yōu)化后的CPU集群的性能還是挺可觀的。拿BigDL來(lái)說(shuō)，MKL + 多線程 + Spark，充分發(fā)揮了分布式集群的優(yōu)勢(shì) 。尤其是在Inference方面，堆CPU的方案在性價(jià)比上很可能是優(yōu)于GPU的，畢竟Nivdia的計(jì)算卡是很昂貴的。

另外，數(shù)據(jù)挖掘以及Information Retrieval等領(lǐng)域中常用的神經(jīng)網(wǎng)絡(luò)結(jié)構(gòu)一般都比較淺，多為稀疏網(wǎng)絡(luò)，也很少用到卷積層。GPU并不十分擅長(zhǎng)處理這樣的網(wǎng)絡(luò)結(jié)構(gòu)。

考慮到實(shí)際的生產(chǎn)環(huán)境，跑在Spark上的BigDL背后有整個(gè)Spark/Hadoop大生態(tài)的支持。配合近期很火的SMACK技術(shù)棧，可以很輕松愉快的構(gòu)建端到端的生產(chǎn)級(jí)別的分布式機(jī)器學(xué)習(xí)流水線。由于沒(méi)有異構(gòu)集群數(shù)據(jù)傳輸?shù)拈_(kāi)銷，從端到端這個(gè)層面來(lái)看，CPU方案的性能反而可能占優(yōu)。

最后，談?wù)効捎眯裕珺igDL項(xiàng)目正在快速的迭代中。語(yǔ)言層面支持Scala/Python。API方面有torch.nn風(fēng)格的Sequenial API，也有TensorFlow風(fēng)格的Graph API，以及正在開(kāi)發(fā)的keras API。Layer庫(kù)也很齊全，自定義Layer也很方便。兼容性方面，BigDL兼容了Caffe/Torch/Keras，以及部分TensorFlow模型。換言之，你可以把用TF/Caffe訓(xùn)練的模型，導(dǎo)入BigDL做Inference。反之，亦可。這是一個(gè)非常有用的Feature。

綜上，BigDL雖然并不主流，但在很多場(chǎng)景下是有成為"大殺器"潛質(zhì)的，包括但不限于：

已有大規(guī)模分布式集群的(如: Hadoop集群)

需要大規(guī)模Inference的，比如：推薦系統(tǒng)、搜索系統(tǒng)、廣告系統(tǒng)

(上下游)依賴Spark/Hadoop生態(tài)的

輕度深度學(xué)習(xí)使用者，如：數(shù)據(jù)研發(fā)工程師/數(shù)據(jù)挖掘工程師

Scala/JVM愛(ài)好者

作者：AlfredXXfiTTshttps://www.zhihu.com/question/54604301/answer/338630738

Analytics Zoo 分析庫(kù)

和 Python 生態(tài)系統(tǒng)中龐大的標(biāo)準(zhǔn)或三方庫(kù)相比，Spark 明顯還處于起步階段。Keras、TensorFlow 和 PyTorch 等大多數(shù)庫(kù)都還不能與 Spark 兼容，因?yàn)樗鼈儾恢С諷park 分布式計(jì)算的底層核心框架。那要如何彌補(bǔ)這一不足呢？這里為大家介紹一個(gè)英特爾開(kāi)發(fā)的分析工具——Analytics Zoo，它提供了一組豐富的高級(jí) API 可以將BigDL、Keras 和 TensorFlow 程序無(wú)縫集成到 Spark 的 pipeline 中；還有幾個(gè)內(nèi)置的深度學(xué)習(xí)模型，可用于對(duì)象檢測(cè)、圖像分類、文本分類等。該庫(kù)還提供端到端的參考用例，如異常檢測(cè)、欺詐檢測(cè)和圖像增強(qiáng)，以將機(jī)器學(xué)習(xí)應(yīng)用于實(shí)際問(wèn)題。

為了幫助大家能更具體、實(shí)際的理解這個(gè)工具的一些功能與用法，下面分享一個(gè)關(guān)于 BigDL 和 Analytics Zoo 的簡(jiǎn)短教程，向大家展示如何使用預(yù)先訓(xùn)練好的模型實(shí)現(xiàn)遷移學(xué)習(xí)，并在 Spark 集群上進(jìn)行訓(xùn)練。

教程實(shí)踐

數(shù)據(jù)集：ResNet-50，包含螞蟻和蜜蜂圖像的小數(shù)據(jù)集來(lái)實(shí)現(xiàn)遷移學(xué)習(xí)。

預(yù)訓(xùn)練模型：可以將給定的圖像在 1000 個(gè)標(biāo)簽中進(jìn)行分類；

模型訓(xùn)練與預(yù)測(cè)：特定用例通過(guò)遷移學(xué)習(xí)重新訓(xùn)練模型，對(duì)包含螞蟻和蜜蜂的訓(xùn)練集進(jìn)行預(yù)測(cè)。BigDL 和 Analytics Zoo 支持在 Spark 的分布式框架上進(jìn)行訓(xùn)練。（注意，最初的 ResNet-50 標(biāo)簽中沒(méi)有“螞蟻”和“蜜蜂”。）

使用 pip 即可安裝 BigDL 和 Analytics Zoo，如下所示：

#for Python3pip3 install BigDLpip3 install analytics-zoo

安裝之后，在開(kāi)始之前先下載 ResNet 50 的預(yù)訓(xùn)練模型、訓(xùn)練與測(cè)試數(shù)據(jù)集。數(shù)據(jù)包需要解壓縮。使用 Analytics Zoo 中的 init_nncontext 函數(shù)導(dǎo)入并初始化 Spark，然后定義預(yù)訓(xùn)練模型、訓(xùn)練與測(cè)試數(shù)據(jù)集的路徑。

import os from bigdl.nn.criterion import * from bigdl.nn.layer import * from bigdl.optim.optimizer import Adam from pyspark.ml import Pipeline from pyspark.ml.evaluation import MulticlassClassificationEvaluator from pyspark.sql.functions import col, udf from pyspark.sql.types import DoubleType, StringType from zoo.common.nncontext import * from zoo.feature.image import * from zoo.pipeline.nnframes import *sc=init_nncontext("TransferLearningBlog")

接下來(lái)，創(chuàng)建 Spark UDF 來(lái)提取文件名稱。標(biāo)簽是通過(guò)檢查文件名稱是否包含關(guān)鍵字“ants”或“bees”來(lái)分配的。使用這兩個(gè) udf，構(gòu)造訓(xùn)練和測(cè)試數(shù)據(jù)集。

# Define udfs to extract filename and generate labels in floats getFileName = udf(lambda row: os.path.basename(row[0]), StringType()) getLabel = udf(lambda row: 1.0 if 'ants' in row[0] else 2.0, DoubleType()) # Construct training dataframe trainingDF = NNImageReader.readImages(train_path, sc, resizeH=300, resizeW=300, image_codec=1) trainingDF = trainingDF.withColumn('filename', getFileName('image')).withColumn('label', getLabel('image')) # Construct validation dataframe validationDF = NNImageReader.readImages(val_path, sc, resizeH=300, resizeW=300, image_codec=1)validationDF=validationDF.withColumn('filename',getFileName('image')).withColumn('label',getLabel('image'))

為了正確構(gòu)建模型，需要對(duì)所有圖像進(jìn)行標(biāo)準(zhǔn)化。Analytics Zoo 有 API 來(lái)操作轉(zhuǎn)換、鏈接等，使后面可以按順序進(jìn)行處理。

如下所示，加載預(yù)訓(xùn)練 ResNet-50 模型

# Create a chained transformer that resizes, crops and normalizes each image in the dataframe transformer = ChainedPreprocessing( [RowToImageFeature(), ImageResize(256, 256), ImageCenterCrop(224, 224), ImageChannelNormalize(123.0, 117.0, 104.0), ImageMatToTensor(), ImageFeatureToTensor()]) # Load pre-trained Resnet-50 that was downloaded earlier and give the column to pick features from preTrainedNNModel = NNModel(Model.loadModel(model_path), transformer) .setFeaturesCol("image") .setPredictionCol("embedding") # Print all layers in Resnet-50 for layer in preTrainedNNModel.model.layers:print(layer.name())

ResNet-50 的最后 5 層是：

res5c_relupool5Viewf42780f5fc1000prob

模型的最后一層的輸出是 2 個(gè)類（螞蟻、蜜蜂），而不是ResNet-50訓(xùn)練的1000個(gè)類。該模型的輸入維數(shù)為 1000，輸出維數(shù)為 2。通過(guò)遷移學(xué)習(xí)，該模型可以在 25 步內(nèi)完成這兩個(gè)新類的訓(xùn)練！這一點(diǎn)也說(shuō)明了遷移學(xué)習(xí)的實(shí)用性。

# Create a last layer with input dimension of 1000 that outputs 2 classes of ants and bees # Epochs are set to 25 and the optimizer is SGD lrModel = Sequential().add(Linear(1000, 2)).add(LogSoftMax()) classifier = NNClassifier(lrModel, ClassNLLCriterion(), SeqToTensor([1000])) .setOptimMethod(SGD(learningrate=0.001, momentum=0.9)) .setBatchSize(4) .setMaxEpoch(25) .setFeaturesCol("embedding") .setCachingSample(False) # Change the last layer in the pipelinepipeline=Pipeline(stages=[preTrainedNNModel,classifier])

現(xiàn)在，開(kāi)始訓(xùn)練和測(cè)試模型。Spark 允許跨多個(gè)集群進(jìn)行更快的訓(xùn)練。

# Train the model and get predictions on the validation set antbeeModel = pipeline.fit(trainingDF) predictionDF = antbeeModel.transform(validationDF).cache() predictionDF.sample(False, 0.1).show() # Evaluate predictions evaluator = MulticlassClassificationEvaluator( labelCol="label", predictionCol="prediction", metricName="accuracy") accuracy = evaluator.evaluate(predictionDF) # expected error should be less than 10%print("TheTestErroris=%g"%(1.0-accuracy))

最后，對(duì)測(cè)試數(shù)據(jù)進(jìn)行分類，顯示圖像。

# Test dataframe testDF = NNImageReader.readImages(test_path, sc, resizeH=300, resizeW=300, image_codec=1) testDF = testDF.withColumn('filename', getFileName('image')).withColumn('label', getLabel('image')) testPredDF = antbeeModel.transform(testDF).cache() row = testPredDF.first().asDict() # showImage function def showImage(row): # Open file plt.imshow(Image.open(row['image'][0][5:])) # Map prediction to class title = 'ants' if row['prediction'] == 1.0 else 'bees' plt.title(title)showImage(row)

測(cè)試數(shù)據(jù)分類結(jié)果的圖像顯示：

如果數(shù)據(jù)集比較大，恰好存儲(chǔ)在 HDFS 中，也可以使用相同的方法，將其擴(kuò)展到更大的集群上。正是 BigDL讓這些大數(shù)據(jù)集的數(shù)據(jù)分析更加快速和高效。除此之外，它還可與 Spark SQL 和結(jié)構(gòu)化數(shù)據(jù)緊密耦合。例如，Kafka 數(shù)據(jù)可以直接傳遞給 BigDL UDF，進(jìn)行實(shí)時(shí)預(yù)測(cè)和分類。