Spark Core相关-1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.SparkSession

object BlockStorageTest {
  def main(args: Array[String]): Unit = {
    // Step 1: Create Spark Session
    val spark = SparkSession.builder()
      .appName("Block Storage Test")
      .master("local[*]")
      .config("spark.local.dir", "D:\\tmp\\spark_data")
      .getOrCreate()

    val sc = spark.sparkContext

    // Step 2: Create an RDD
    val data = 1 to 1000 // Sample data
    val rdd = sc.parallelize(data, numSlices = 4)

    // Step 3: Persist the RDD to memory and disk
    rdd.persist(StorageLevel.DISK_ONLY)

    // Step 4: Trigger an action to store the blocks
    rdd.count() // Action to compute and store the blocks

    // Step 5: Log the disk storage location
    println("Disk storage location: " + sc.getConf.get("spark.local.dir"))

    // Keep the application alive to inspect the disk manually
    Thread.sleep(60000)

    // Stop SparkSession
    spark.stop()
  }
}

1

private[spark] sealed trait MapOutputTrackerMessage

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

private[spark] case class GetMapOutputStatuses(shuffleId: Int)
  extends MapOutputTrackerMessage
private[spark] case class GetMapAndMergeResultStatuses(shuffleId: Int)
  extends MapOutputTrackerMessage
private[spark] case class GetShufflePushMergerLocations(shuffleId: Int)
  extends MapOutputTrackerMessage
private[spark] case object StopMapOutputTracker extends MapOutputTrackerMessage

private[spark] sealed trait MapOutputTrackerMasterMessage
private[spark] case class GetMapOutputMessage(shuffleId: Int,
  context: RpcCallContext) extends MapOutputTrackerMasterMessage
private[spark] case class GetMapAndMergeOutputMessage(shuffleId: Int,
  context: RpcCallContext) extends MapOutputTrackerMasterMessage
private[spark] case class GetShufflePushMergersMessage(shuffleId: Int,
  context: RpcCallContext) extends MapOutputTrackerMasterMessage
private[spark] case class MapSizesByExecutorId(
  iter: Iterator[(BlockManagerId, 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

  def initialize(): Unit = {
    val jobsTab = new JobsTab(this, store)
    attachTab(jobsTab)
    val stagesTab = new StagesTab(this, store)
    attachTab(stagesTab)
    attachTab(new StorageTab(this, store))
    attachTab(new EnvironmentTab(this, store))
    attachTab(new ExecutorsTab(this))
    addStaticHandler(SparkUI.STATIC_RESOURCE_DIR)
    attachHandler(createRedirectHandler("/", "/jobs/", basePath = basePath))
    attachHandler(ApiRootResource.getServletHandler(this))
    if (sc.map(_.conf.get(UI_PROMETHEUS_ENABLED)).getOrElse(false)) {
      attachHandler(PrometheusResource.getServletHandler(this))
    }

    // These should be POST only, but, the YARN AM proxy won't proxy POSTs
    attachHandler(createRedirectHandler(
      "/jobs/job/kill", "/jobs/", jobsTab.handleKillRequest, httpMethods = Set("GET", "POST")))
    attachHandler(createRedirectHandler(
      "/stages/stage/kill", "/stages/", stagesTab.handleKillRequest,
      httpMethods = Set("GET", "POST")))
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

  def attachPage(page: WebUIPage): Unit = {
    val pagePath = "/" + page.prefix
    val renderHandler = createServletHandler(pagePath,
      (request: HttpServletRequest) => page.render(request), conf, basePath)
    val renderJsonHandler = createServletHandler(pagePath.stripSuffix("/") + "/json",
      (request: HttpServletRequest) => page.renderJson(request), conf, basePath)
    attachHandler(renderHandler)
    attachHandler(renderJsonHandler)
    val handlers = pageToHandlers.getOrElseUpdate(page, ArrayBuffer[ServletContextHandler]())
    handlers += renderHandler
    handlers += renderJsonHandler
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

  def createServletHandler(
      path: String,
      servlet: HttpServlet,
      basePath: String): ServletContextHandler = {
    val prefixedPath = if (basePath == "" && path == "/") {
      path
    } else {
      (basePath + path).stripSuffix("/")
    }
    val contextHandler = new ServletContextHandler
    val holder = new ServletHolder(servlet)
    contextHandler.setContextPath(prefixedPath)
    contextHandler.addServlet(holder, "/")
    contextHandler
  }

1
2
3
4
5
6

class MyTab(parent: SparkUI, store: AppStatusStore)
  extends SparkUITab(parent, "my_tab") {

  attachPage(new MyStoragePage(this, store))

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

  def go(): Unit = {
    val spark = SparkSession.builder()
      .appName("Block Storage Test")
      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    val ui: SparkUI = sc.ui.getOrElse(throw new RuntimeException("Not support web ui."))
    val store = ui.store
    ui.attachTab(new MyTab(ui, store))
    Thread.sleep(Long.MaxValue)
  }

1
2
3

    ui.attachHandler()
    ui.addStaticHandler()
    ui.attachPage()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import org.apache.spark.{SparkConf, SparkContext}

object RpcServer {

  class HelloWorldEndpoint(val rpcEnv: RpcEnv) extends RpcEndpoint {
    override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
      case "Hello" =>
        println("Received 'Hello' message")
        context.reply("Hello from the server!")
      case _ =>
        context.reply("Unknown message")
    }

    override def receive: PartialFunction[Any, Unit] = {
      case msg => println(s"Received: $msg")
    }

    override def onStop(): Unit = {
      println("Stopping the server endpoint")
    }
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Spark RPC Hello World").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val rpcEnv: RpcEnv = RpcEnv.create("HelloWorldServer", "localhost",
      9999, conf, sc.env.securityManager)

    // Register the HelloWorld endpoint
    val helloEndpoint = rpcEnv.setupEndpoint("helloEndpoint", new HelloWorldEndpoint(rpcEnv))

    println("Server is running and waiting for messages...")
    Thread.sleep(Long.MaxValue)  // Keep the server running
    rpcEnv.stop(helloEndpoint)  // Shutdown the server
    sc.stop()  // Stop SparkContext
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

import org.apache.spark.{SparkConf, SparkContext}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

object RpcClient {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Spark RPC Hello World Client").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val rpcEnv: RpcEnv = RpcEnv.create("HelloWorldClient", "localhost",
      9999, conf, sc.env.securityManager)

    // Connect to the server endpoint
    val address = new RpcAddress("localhost", 9999)
    val helloEndpoint = rpcEnv.setupEndpointRef(address, "helloEndpoint")

    // Send a message to the server
    helloEndpoint.ask[String]("Hello").onComplete {
      case Success(response) =>
        println(s"Received from server: $response")
      case Failure(e) =>
        println(s"Failed to receive response from server: ${e.getMessage}")
    }

    // Let the client wait for the response
    Thread.sleep(2000)

    sc.stop()
  }
}

1

val outboxes = new ConcurrentHashMap[RpcAddress, Outbox]()

1

val messages = new java.util.LinkedList[OutboxMessage]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    // Messages sent and received locally
    case ExecutorRegistered(executorId) =>
      executorLastSeen(executorId) = clock.getTimeMillis()
      context.reply(true)
    case ExecutorRemoved(executorId) =>
      executorLastSeen.remove(executorId)
      context.reply(true)
    case TaskSchedulerIsSet =>
      scheduler = sc.taskScheduler
      context.reply(true)
    case ExpireDeadHosts =>
      expireDeadHosts()
      context.reply(true)
...
...
}	  

1
2

val endpoints: ConcurrentMap[String, MessageLoop] =
    new ConcurrentHashMap[String, MessageLoop]