import spark.SparkContextimport SparkContext._//导入spark开发包
object SparkTest {  def main( args: Array[String]) {    
if (args.length == 0) {      
System.err.println("Usage: SparkTest <host> [<slices>]")      
System.exit(1)    
}   
 val spark = new SparkContext(args(0), "SparkTest")//创建SparkContext对象，告诉spark如何访问spark集群    
val slices = if (args.length > 1) args(1).toInt else 2//定义数据集分为几块    
val myFile = spark.textFile("test.txt")//创建一个分布式数据集   
 val counts = myFile.flatMap(line => line.split(" ")//文件中每一行单词以空格分隔，转换的过程                        
	.map(word => (word, 1))//数据集转换成（word,1）键值对，也是数据集转换的过程                        
	.reduceByKey(_ + _)//统计单词个数，动作的过程     
	counts.saveAsTextFile("out.txt")//输出文件   
} 
}
 SparkTest.main(args)//设置main函数参数

1、用Scala语言实现Logistic Regression Classifier的完整代码：

import java.util.Random  
import java.lang.Math  
import scala.collection.mutable.HashMap  import scala.io.Source  
import org.apache.spark.SparkContext //创建SparkContext对象，告诉spark如何访问spark集群
import org.apache.spark.rdd.RDD;  //RDDs
import org.apache.spark.util.Vector  //spark容器
import org.apache.spark.broadcast.Broadcast  //广播变量
import spark.ml.utils.SparserVector  
object SparseLR {  //定义一个类：LR分类器  
val labelNum = 2; // 类别数    
val dimNum = 124; // 维度   
val iteration = 10; // 迭代次数    
val alpha = 0.1 // 迭代步长   
val lambda = 0.1    
val rand = new Random(42)  //取一随机数  
var w = Vector(dimNum, _ => rand.nextDouble) //用随机数初始化参数 ，w为一124维的容器变量    
//定义一个数据点    
case class DataPoint(x: SparserVector, y: Int) //整形y，SparserVector类型的x(124维)  
// 解析一个读入的训练样本，构造DataPoint结构    
def parsePoint(line: String): DataPoint = {  //line为传入的训练样本数据    
var features = new SparserVector(dimNum)      
val fields = line.split(" ") //transformation过程，将原RDDs数据集转换为以空格分割后的数据集     
val y = fields(0).toInt      
fields.filter(_.contains(":")).foreach(f => {        
val feature = f.split(":")        
features.insert(feature(0).toInt, feature(1).toDouble)      
})     
DataPoint(features, y)    
}   

//读样本文件，构造训练数据   
def genearteDataPoints(filename: String): Array[DataPoint] = {      
val dataPoints = Source.fromFile(filename).getLines.map(parsePoint).toArray     
dataPoints    
}   

//定义sigmod函数  
def sigmod(x: SparserVector): Double = {      
val features = x.elements      
val s = features.keySet.map(k => w(k) * features.get(k).get).reduce(_ + _)//权值和训练数据乘积之和     
1 / (1 + Math.exp(-s))  //逻辑函数  
}     

// train函数，根据样本训练参数   
def train(sc: SparkContext, dataPoints: RDD[DataPoint]) {      
val sampleNum = dataPoints.count        //开始迭代      
for (i <- 0 until iteration) {        
val g = dataPoints.map(p => p.x * (sigmod(p.x) - p.y)).reduce(_ + _) + lambda * w        
w -= alpha * g  //牛顿－拉斐森(Newton-Raphson)方法进行迭代求解。  	
println("iteration " + i + ": g = " + g)  	
println("iteration " + i + ": w = " + w)      
}    
}    

//根据训练出的参数进行预测   
def predict(dataPoints: RDD[DataPoint]): Double = {      
val correct = dataPoints.map(p => {       
val label = if (sigmod(p.x) > 0.5) 1 else 0       
if (label == p.y) 1 else 0      
}).reduce(_ + _)        
(correct * 1.0) / dataPoints.count  
  }    
  
  //main函数  
  def main(args: Array[String]): Unit = {      
  val trainfile = "data/a8a.train";        
  //val sc = new SparkContext(args(0), "LR")     
  val sc = new SparkContext("local", "LR")      
  val trainset = sc.textFile(trainfile, 2).map(parsePoint).cache       
  train(sc, trainset)  //训练样本，得到最优参数      
  val testfile = "data/a8a.test";     
  val testset = sc.textFile(testfile, 2).map(parsePoint).cache      
  val accuracy = predict(testset)  //测试数据，得到分类结果     
  println(accuracy)    
  }  
  }  

2、Spark的LR分类器

package spark.examples
Import scala.io.Source
import java.util.Random
import scala.math.exp
import spark.util.Vector
import spark._
*******************************************************************
object SparkLR {
  val N = 10000  // Number of data points
  val D = 10   // Numer of dimensions
  val R = 0.7  // Scaling factor
  val ITERATIONS = 5
  val rand = new Random(42)
  case class DataPoint(x: Vector, y: Double)
  case class DataPoint1(x: Vector)
********************************************************************
  def generateData = {     //自己构建了训练数据
    def generatePoint(i: Int) = {
      val y = if(i % 2 == 0) -1 else 1
      val x = Vector(D, _ => rand.nextGaussian + y * R)
      DataPoint(x, y)
    }
    Array.tabulate(N)(generatePoint)
  }
*************************************************************************
//fang该部分是对上面的数据结构改进，方便从文件中读入数据
def generateData = {     //自己构建了训练数据
Val src=Source.fromFile(“/home/jay/file01”)
Val iter=src.getLines()
    def generatePoint(i: Int) = {
      val y = if(i % 2 == 0) -1 else 1
      val x = Vector(D, iter.next().split(“ ”).map(w=>w.toDouble))
      DataPoint(x, y)
    }
Array.tabulate(N)(generatePoint)
  }

def generateData = {     //自己构建测试数据
Val src1=Source.fromFile(“/home/jay/file02”)
Val iter1=src1.getLines()
    def generatePoint1(i: Int) = {
      val x = Vector(D, iter1.next().split(“ ”).map(w=>w.toDouble))
      DataPoint1(x)
    }
Array.tabulate(N)(generatePoint1)
  }
//fang
*******************************************************************************
  def main(args: Array[String]) {
    Val conf = new SparkConf().setAppName(“SparkLR”)
    val sc = new SparkContext(conf)
    val numSlices = if (args.length > 1) args(1).toInt else 2
val points = sc.parallelize(generateData, numSlices).cache()
val points1 = sc.parallelize(generateData1, numSlices).cache()
    // Initialize w to a random value
    var w = Vector(D, _ => 2 * rand.nextDouble - 1)
    println("Initial w: " + w)

    for (i <- 1 to ITERATIONS) {//运用牛顿迭代法实现LR分类器
      println("On iteration " + i)
      val gradient = points.map { p =>
        (1 / (1 + exp(-p.y * (w dot p.x))) - 1) * p.y * p.x
      }.reduce(_ + _)
      w -= gradient
    }
println("Final w: " + w)
Val y=points1.map{p=>(1/(1+exp(w dot p.x)))}
var Y = Vector(N,  y.take(N)
Println(“result: ”, + Y)
    System.exit(0)
  }
}

训练数据模型：

xxx   xxx  xxx  xxx  xxx  xxx  xxx  xxx  xxx  xxx  yyy

版权声明：本文为博主原创文章，未经博主允许不得转载。