06-CCTC2016-聚效廣告劉憶智-Beyond MLLibScale up Advanced Machine Learning on Spark

ScaleupAdvancedMachineLearningonSparkQihoo360/MVADDMLCmemberMXNetcommitterAlgorithmsinSparkML?ClassificationandRegression?Linearmodels?NaiveBayescisionTreesnsemblesofTrees?IsotonicRegressionCollaborativeFiltering?ALS?……?ClusteringLatentDirichletAllocation?……?DimensionalityReduction?……wwwwComputationalModelwwnAdvancedMLAlgorithmsXGBoostisanoptimizeddistributedgradientexibleandportableboostingXGBoostisanoptimizeddistributedgradientexibleandportableMXNetisadeeplearningframeworkdesignedforbothe?ciencyand?exibilityp(cat)p(dog)p(cat)p(dog)efinethenetworktionalcostlayersemorylimitTrainDeepNetworkonMXNet(declaratively)nvaldataSymbolVariable"data")valfc1=Symbol.FullyConnected(name="fc1")(Map("data"->data,valdataSymbolVariable"data")valact1=Symbol.Activation(name="relu1")(Map("data"->fc1,"act_type"->"relu"))valfc2=Symbol.FullyConnected(name="fc2")(Map("data"->act1,"num_hidden"->64))valact2=Symbol.Activation(name="relu2")(Map("data"->fc2,"act_type"->"relu"))valfc3=Symbol.FullyConnected(name="fc3")(Map("data"->act2,"num_hidden"->10))valmlp=Symbol.SoftmaxOutput(name="sm")(Map("data"->fc3))valmodelFeedForwardnewBuildermlp)valmodelFeedForwardnewBuildermlp).setContext(Context.cpu()).setNumEpoch(10).setOptimizer(newSGD(learningRate=0.1f,momentum=0.9f,wd=0.0001f)).setTrainData(trainDataIter).setEvalData(valDataIter).build()valprobArrays=model.predict(valDataIter)//inthiscase,wedonothavemultipleoutputserrequire(probArrays.lengthervalprob=probArrays(0)//getpredictedlabelsvalpy=NDArray.argmaxChannel(prob)//dealwithpredictedlabelspyMXNetMXNetCompareCompareMXNettoothersonasingleforward-backwardperformanceiceiceGPUSupporticeherevalweight=NDArray.empty(Shape(3,2),Contexticehereweight-=eta*(grad+lambda*weight);valweightOnCpuweightcopyToContextcpu)CPUandGPUtemplate<typenamexpu>voidUpdateSGD(Tensor<xpu,2>weight,constTensor<xpu,2>&grad,floateta,floatlambda){weight-=eta*(grad+lambda*weight);}GPUSupport:mshadow/dmlc/mshadow?Efficient:alltheexpressionyouwritewillbelazilyevaluatedandcompiledintooptimizedcode.?Notemporalmemoryallocationwillhappenforexpressionyouwrite.?mshadowwillgeneratespecifickernelforeveryexpressionyouwriteincompiletime.?Deviceinvariant:youcanwriteonecodeanditwillrunonbothCPUandGPU.?Simple:mshadowallowsyoutowritemachinelearningcodeusingexpressions.?Whitebox:putafloat*intotheTensorstructandtakethebenefitofthepackage,nomemoryallocationishappenedunlessexplicitlycalled.?Lightweightlibrary:lightamountofcodetosupportfrequentlyusedfunctionsinmachinelearning.?Extendable:usercanwritesimplefunctionsthatplugsintomshadowandrunonGPU/CPU,noexperienceinCUDAisrequired.?MultiGPUandDistributedML:mshadow-psinterfaceallowsusertowriteefficientMultiGPUanddistributedprogramsinanunifiedway.dserversdserversvalenvs=Map("DMLC_ROLE"->role,"DMLC_PS_ROOT_URI"->schedulerHost,"DMLC_PS_ROOT_PORT"->schedulerPort,"DMLC_NUM_WORKER"->numWorker,vs"DMLC_NUM_SERVER"vsif(role=="server"||role=="scheduler"){//scheduler&serverKVStoreServer.start()}else{ronousupdatevalkv=KVStore.create("dist_sync")model.fit(trainData=train,kvStore=kv)tsyncIterationKVStore.create("dist_async")IterationKVStore.create("dist_async")AsynchronousExecutionk?Sparkhasbecomethedefactoprocessing.?CombineETLwithmachinelearningpipeline..setNumServer(2)setupserver&.setNumWorker(4)workernumber.setLabelName.setNumServer(2)setupserver&.setNumWorker(4)workernumber.setLabelName("softmax_label")mlpwillbeserializedkAPIvalmxnet=newMXNet().setBatchSize(128).setContext(Context.gpu(0))toworkers.setDimension(Shape(784)).setNetwork(mlp).setExecutorClasspath(classpaths).setOptimizer(newSGD(learningRate=0.01f,momentum=0.9f,wd=0.00001f)valmodel=mxnet.train(trainData)model.save(modelPath)alpredictionsmodelpredicttestDataServerComponentsonSparkmapRange(0,numServer)0MXNetGPUMXNetCPUSparkMLTrain660,000sampleson5EC2nodesengineissingleton?CPU&GPUallocationoncluster?Whatifthere’snoenoughresourcesincluster?execs=[symbol.bind(mx.gpuexecs=[symbol.bind(mx.gpu(i))foriinrange(ngpu)]%w-=learning_rate*gradkvstore.set_updater(…)%iteratingondatafordbatchintrain_iter:%iteratingonGPUsforiinrange(ngpu):%readadatapartitioncopy_data_slice(dbatch,execs[i])%pulltheparametersforkeyinupdate_keys:kvstore.pull(key,execs[i].weight_array[key])%computethegradientexecs[i].forward(is_train=True)execs[i].backward()%pushthegradientforkeyinupdate_keys:kvstore.push(key,execs[i].grad_array[key])UGPUGPU1GPUGPU2GPUGPU3XGBoostSingleMachinePerformanceoostonSparkgonSingleMachinevalparams=newmutable.HashMap[String,Any]()params+="eta"->1.0params+="max_depth"->2params+="silent"->1params+="objective"->"binary:logistic"valwatches=newmutable.HashMap[String,DMatrix]watches+="train"->trainMaxwatches+="test"->testMaxvalround=2//trainamodelvalbooster=XGBoost.train(trainMax,params.toMap,round,watches.toMap)valpredicts=booster.predict(testMax)delpredicttestSetdelpredicttestSetwrapwrapinRDD=MLUtils.loadLibSVMFile(sc,inputTrainPath).repartition(args(1).toInt)valxgboostModel=XGBoost.train(trainRDD,paramMap,numRound,numWorkers)//testSetisanRDDcontainingtestsetdatarepresentedas//org.apache.spark.mllib.regression.LabeledPointvaltestSet=MLUtils.loadLibSVMFile(sc,inputTestPath)//localprediction//importmethodsinDataUtilstoconvertIterator[org.apache.spark.mllib.regression.LabeledPoint]//toIterator[ml.dmlc.xgboost4j.LabeledPoint]inautomaticimportDataUtils._xgboostModel.predict(newDMatrix(testSet.collect().iterator)?Portable:rabitislightweightandrunseverywhere.?Rabitisalibraryinsteadofaframework.?Rabitonlyreliesonamechanismtostartprogram,whichwasprovidedbymostframework.?ScalableandFlexible:rabitrunsfast?RabitprogramuseAllreducetocommunicate,anddonotsufferthecostbetweeniterationsofMapReduceabstraction.?Programscancallrabitfunctionsinanyorder,asopposedtoframeworkswherecallbacksareofferedandcalle