因为MLlib不支持稀疏输入,所以我在Spark集群上运行了支持稀疏输入格式的下面代码。设置如下:
- 5个节点,每个节点有8个核心(每个节点的所有CPU都是100%,当运行代码时,用户模型占98%)。
- 输入:存储在HDFS中的10,000,000+个实例和600,000+维度。
import java.util.Random
import scala.collection.mutable.HashMap
import scala.io.Source
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.util.Vector
import java.lang.Math
import org.apache.spark.broadcast.Broadcast
object SparseLR {
val lableNum = 1
val dimNum = 632918
val iteration = 10
val alpha = 0.1
val lambda = 0.1
val rand = new Random(42)
var w = Vector(dimNum, _=> rand.nextDouble)
class SparserVector {
var elements = new HashMap[Int, Double]
def insert(index: Int, value: Double){
elements += index -> value;
}
def *(scale: Double): Vector = {
var x = new Array[Double](dimNum)
elements.keySet.foreach(k => x(k) = scale * elements.get(k).get)
Vector(x)
}
}
case class DataPoint(x: SparserVector, y: Int)
def parsePoint(line: String): DataPoint = {
var features = new SparserVector
val fields = line.split("\t")
//println("fields:" + fields(0))
val y = fields(0).toInt
fields.filter(_.contains(":")).foreach( f => {
val feature = f.split(":")
features.insert(feature(0).toInt, feature(1).toDouble)
})
return DataPoint(features, y)
}
def gradient(p: DataPoint, w: Broadcast[Vector]) : Vector = {
def h(w: Broadcast[Vector], x: SparserVector): Double = {
val wb = w.value
val features = x.elements
val s = features.keySet.map(k => features.get(k).get * wb(k)).reduce(_ + _)
1 / (1 + Math.exp(-p.y * s))
}
p.x * (-(1 - p.y *h(w, p.x)))
}
def train(sc: SparkContext, dataPoints: RDD[DataPoint]) {
//val sampleNum = dataPoints.count
val sampleNum = 11680250
for(i <- 0 until iteration) {
val wb = sc.broadcast(w)
val g = (dataPoints.map(p => gradient(p, wb)).reduce(_ + _) + lambda * wb.value) /sampleNum
w -= alpha * g
println("iteration " + i + ": g = " + g)
}
}
def main(args : Array[String]): Unit = {
System.setProperty("spark.executor.memory", "15g")
System.setProperty("spark.default.parallelism", "32");
val sc = new SparkContext("spark://xxx:12036", "LR", "/xxx/spark", List("xxx_2.9.3-1.0.jar"))
val lines = sc.textFile("hdfs:xxx/xxx.txt", 32)
val trainset = lines.map(parsePoint _).cache()
train(sc, trainset)
}
}
有人能帮我吗?谢谢!