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

目录

pom.xml

Task：封装数据和逻辑

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

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

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

测试

pom.xml

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.0.0</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <!--该插件用于把Scala代码编译成为class文件-->
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.2</version>
            <executions>
                <execution>
                    <!--声明绑定到maven的compile阶段-->
                    <goals>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.1.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

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