spark学习笔记（一）——模拟分布式计算

目录

pom.xml

Task：封装数据和逻辑

Subtask：分享Task的数据和逻辑进行计算

driver：模拟client，分配计算任务

executer：接收driver传输的数据执行计算

测试

pom.xml

    
        org.apache.spark
        spark-core_2.12
        3.0.0
    


    
        
        
            net.alchim31.maven
            scala-maven-plugin
            3.2.2
            
                
                    
                    
                        testCompile
                    
                
            
        
        
            org.apache.maven.plugins
            maven-assembly-plugin
            3.1.0
            
                
                    jar-with-dependencies
                
            
            
                
                    make-assembly
                    package
                    
                        single
                    
                
            
        
    

Task：封装数据和逻辑

class Task extends Serializable {

  //数据
  val datas = List(2,4,6,8)
  //逻辑：x2
  val logic = ( num:Int) => {num * 2}
}

Subtask：分享Task的数据和逻辑进行计算

class SubTask extends Serializable {
  
  var datas: List[Int] = _
  var logic: (Int) => Int = _

  //计算
  def compute(): List[Int] = {
    datas.map(logic)
  }
}

driver：模拟client，分配计算任务

object driver {
  def main(args: Array[String]): Unit = {
    //连接服务器
    val client1 = new Socket("localhost",9999)
    val client2 = new Socket("localhost",8888)

    val task = new Task()
    val subTask1 = new SubTask()
    //取Task逻辑
    subTask1.logic = task.logic
    //取Task的数据
    subTask1.datas = task.datas.take(2)

    val out1: OutputStream = client1.getOutputStream
    val objOut1 = new ObjectOutputStream(out1)
    //写人数据、刷新、关闭
    objOut1.writeObject(subTask1)
    objOut1.flush()
    objOut1.close()
    client1.close()
    println("客服端[9999]发送数据完成！")


    val subTask2 = new SubTask()
    //取Task逻辑
    subTask2.logic = task.logic
    //取Task的数据
    subTask2.datas = task.datas.takeRight(2)

    val out2: OutputStream = client2.getOutputStream
    val objOut2 = new ObjectOutputStream(out2)
    //写人数据、刷新、关闭
    objOut2.writeObject(subTask2)
    objOut2.flush()
    objOut2.close()
    client2.close()
    println("客服端[8888]发送数据完成！")
  }
}

executer：接收driver传输的数据执行计算

executer1

object Executer {
  def main(args: Array[String]): Unit = {

    //启动服务器接收数据
    val server = new ServerSocket(9999)
    println("服务器启动，等待接收数据！")

    //等待客服端连接
    val client: Socket = server.accept()
    val in: InputStream = client.getInputStream

    val objIn = new ObjectInputStream(in)
    val task: SubTask = objIn.readObject().asInstanceOf[SubTask]
    val ints: List[Int] = task.compute()
    println("计算节点[9999]计算的结果为：" + ints)
    objIn.close()
    client.close()
    server.close()
  }
}

executer2

object Executer2 {
  def main(args: Array[String]): Unit = {

    //启动服务器接收数据
    val server = new ServerSocket(8888)
    println("服务器启动，等待接收数据！")

    //等待客服端连接
    val client: Socket = server.accept()
    val in: InputStream = client.getInputStream

    val objIn = new ObjectInputStream(in)
    val task: SubTask = objIn.readObject().asInstanceOf[SubTask]
    val ints: List[Int] = task.compute()
    println("计算节点[8888]计算的结果为：" + ints)
    objIn.close()
    client.close()
    server.close()
  }
}

测试

启动顺序executer1----->executer2------>driver

 

查看executer1和executer2