大數據+深度學習下，BigDL框架的利弊與應用教程

【導讀】你能利用現有的 Spark 集群構建深度學習模型嗎？如何分析存儲在 HDFS、Hive 和 HBase 中 tb 級的數據嗎？企業想用深度學習模型，可是要考慮的問題又很多，怎么破？這篇文章中，我們將給大家講講大數據+深度學習下，BigDL 框架的利弊與應用教程，為什么有了 TF、PyTorch，還是會考慮用 BigDL？

為什么要講 BigDL？

這幾年，曾被稱為 “3S”，因其簡單、快速并支持深度學習的 Apache Spark 非常流行。許多公司利用 Hadoop 和 Spark 環境來構建強大的數據處理 pipeline，對分布式集群上的大量數據進行預處理，并從中挖掘出業務提升的新觀點?，F在許多公司都希望能利用深度學習的模型幫助自己進一步改善業務。雖然深度學習模型的性能在不斷提高，但是想要在現有的平臺上部署新技術也還有很多問題需要權衡，比如：

（1）如果用深度學習的方法，還可以利用原有的 pipeline 嗎？

（2）當深度學習遇到大規模數據集時，“大規模深度學習”如何能保證其有效性？

（3）基于現有的 Spark / Hadoop 集群是否可以用？

為什么要權衡這些問題其實不難理解，我們需要保持一致的環境，避免大型數據集跨不同集群之間的傳遞。此外，從現有的基礎設施中移動專有數據集也有安全風險與隱患。早期時解決這些問題的方法是在 Spark 上直接加入深度學習框架，但并不能保證保持它們之間的一致性，因此，后來產生了基于 Spark 的 BigDL 平臺，其繼承了 3S 的主要特點：簡單、快速、支持深度學學習。

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

BigDL 是一個分布式的深度學習框架，在大數據分析領域發展迅速，并且也是一個開源的框架。BigDL 有很多特點，比如：與 Spark 和 Hadoop 生態系統進行了完整集成，具有可拓展性等很多重要的功能?？筛鶕祿笮≡谌我饧褐杏柧毮Ｐ?、支持構建端到端的大數據分析與深度學習等 pipeline、可執行數據并行分布式訓練，實現高可擴展性。BigDL 用戶可在 Spark 和大數據平臺上構建了大量數據分析與深度學習的應用，如視覺相似性、參數同步、比例縮放等。

深度學習應用程序可以編寫為標準的 spark 庫。這些 Spark 框架中統一的庫可以讀取大量數據。此外，它還支持 Numpy、Scipy、NLTK、Pandas 等 Python 庫；與 TensorBoard 集成用于可視化分析；支持加載現有的 Torch 模型。企業客戶使用 BigDL 和Spark 還有一個重要的原因，相比 TensorFlow，BigDL 不僅更快，通過并行計算它能夠更快地重新訓練模型。

分享一位網友對 BigDL 的總結：

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

雖然業界普遍不看好CPU跑深度學習，但實際上還是有需求的。比如，現有Hadoop集群的公司，復用現有集群來跑深度學習是最經濟的方案。

并且，充分優化后的CPU集群的性能還是挺可觀的。拿BigDL來說，MKL + 多線程 + Spark，充分發揮了分布式集群的優勢 。尤其是在Inference方面，堆CPU的方案在性價比上很可能是優于GPU的，畢竟Nivdia的計算卡是很昂貴的。

另外，數據挖掘以及Information Retrieval等領域中常用的神經網絡結構一般都比較淺，多為稀疏網絡，也很少用到卷積層。GPU并不十分擅長處理這樣的網絡結構。

考慮到實際的生產環境，跑在Spark上的BigDL背后有整個Spark/Hadoop大生態的支持。配合近期很火的SMACK技術棧，可以很輕松愉快的構建端到端的生產級別的分布式機器學習流水線。由于沒有異構集群數據傳輸的開銷，從端到端這個層面來看，CPU方案的性能反而可能占優。

最后，談談可用性，BigDL項目正在快速的迭代中。語言層面支持Scala/Python。API方面有torch.nn風格的Sequenial API，也有TensorFlow風格的Graph API，以及正在開發的keras API。Layer庫也很齊全，自定義Layer也很方便。兼容性方面，BigDL兼容了Caffe/Torch/Keras，以及部分TensorFlow模型。換言之，你可以把用TF/Caffe訓練的模型，導入BigDL做Inference。反之，亦可。這是一個非常有用的Feature。

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

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

需要大規模Inference的，比如：推薦系統、搜索系統、廣告系統

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

輕度深度學習使用者，如：數據研發工程師/數據挖掘工程師

Scala/JVM愛好者

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

Analytics Zoo 分析庫

和 Python 生態系統中龐大的標準或三方庫相比，Spark 明顯還處于起步階段。Keras、TensorFlow 和 PyTorch 等大多數庫都還不能與 Spark 兼容，因為它們不支持Spark 分布式計算的底層核心框架。那要如何彌補這一不足呢？這里為大家介紹一個英特爾開發的分析工具——Analytics Zoo，它提供了一組豐富的高級 API 可以將BigDL、Keras 和 TensorFlow 程序無縫集成到 Spark 的 pipeline 中；還有幾個內置的深度學習模型，可用于對象檢測、圖像分類、文本分類等。該庫還提供端到端的參考用例，如異常檢測、欺詐檢測和圖像增強，以將機器學習應用于實際問題。

為了幫助大家能更具體、實際的理解這個工具的一些功能與用法，下面分享一個關于 BigDL 和 Analytics Zoo 的簡短教程，向大家展示如何使用預先訓練好的模型實現遷移學習，并在 Spark 集群上進行訓練。

教程實踐

數據集：ResNet-50，包含螞蟻和蜜蜂圖像的小數據集來實現遷移學習。

預訓練模型：可以將給定的圖像在 1000 個標簽中進行分類；

模型訓練與預測：特定用例通過遷移學習重新訓練模型，對包含螞蟻和蜜蜂的訓練集進行預測。BigDL 和 Analytics Zoo 支持在 Spark 的分布式框架上進行訓練。（注意，最初的 ResNet-50 標簽中沒有“螞蟻”和“蜜蜂”。）

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

#for Python3pip3 install BigDLpip3 install analytics-zoo

安裝之后，在開始之前先下載 ResNet 50 的預訓練模型、訓練與測試數據集。數據包需要解壓縮。使用 Analytics Zoo 中的 init_nncontext 函數導入并初始化 Spark，然后定義預訓練模型、訓練與測試數據集的路徑。

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")

接下來，創建 Spark UDF 來提取文件名稱。標簽是通過檢查文件名稱是否包含關鍵字“ants”或“bees”來分配的。使用這兩個 udf，構造訓練和測試數據集。

# 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'))

為了正確構建模型，需要對所有圖像進行標準化。Analytics Zoo 有 API 來操作轉換、鏈接等，使后面可以按順序進行處理。

如下所示，加載預訓練 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 個類（螞蟻、蜜蜂），而不是ResNet-50訓練的1000個類。該模型的輸入維數為 1000，輸出維數為 2。通過遷移學習，該模型可以在 25 步內完成這兩個新類的訓練！這一點也說明了遷移學習的實用性。

# 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])

現在，開始訓練和測試模型。Spark 允許跨多個集群進行更快的訓練。

# 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))

最后，對測試數據進行分類，顯示圖像。

# 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)

測試數據分類結果的圖像顯示：

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