Spark集群 + Akka + Kafka + Scala 开发(3) : 开发一个Akka + Spark的应用

前言

akka是什么？

akka的作用

akka的运行模式和用例

WEB-INF/lib

本文中的用例

as an application

项目结构和文件说明

说明

文件结构

AkkaSampleApp    # 项目目录
|-- build.bat    # build文件
|-- src
|-- main
|-- resources
|-- application.conf   # Akka Server应用的配置文件
|-- client.conf        # Akka Client应用的配置文件
|-- scala
|-- ClientActor.scala       # Akka Client的Actor：提供了一种调用Server Actor的方式。
|-- ClientApp.scala         # Akka Client应用
|-- ProductionReaper.scala  # Akka Shutdown pattern的实现者
|-- Reaper.scala            # Akka Shutdown pattern的Reaper抽象类
|-- ServerActor.scala       # Akka Server的Actor，提供一个求1到n的MapReduce计算。使用了Spark。
|-- ServerApp.scala         # Akka Server应用

构建工程目录

mkdir AkkaSampleApp
mkdir -p /AkkaSampleApp/src/main/resources
mkdir -p /AkkaSampleApp/src/main/scala

代码

build.sbt

name := "akka-sample-app"

version := "1.0"

scalaVersion := "2.11.8"

scalacOptions += "-feature"
scalacOptions += "-deprecation"
scalacOptions += "-language:postfixOps"

libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-actor" % "2.4.10",
"com.typesafe.akka" %% "akka-remote" % "2.4.10",
"org.apache.spark" %% "spark-core" % "2.0.0"
)

resolvers += "Akka Snapshots" at "http://repo.akka.io/snapshots/"

application.conf

akka {
#loglevel = "DEBUG"
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2552
}
#log-sent-messages = on
#log-received-messages = on
}
}

cient.conf

akka {
#loglevel = "DEBUG"
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 0
}
#log-sent-messages = on
#log-received-messages = on
}
}

port = 0

ClientActor.scala

import akka.actor._
import akka.event.Logging

class ClientActor(serverPath: String) extends Actor {
val log = Logging(context.system, this)
val serverActor = context.actorSelection(serverPath)

def receive = {
case msg: String =>
log.info(s"ClientActor received message '$msg'")
serverActor ! 10000L
}
}

ClientApp.scala

import com.typesafe.config.ConfigFactory
import akka.actor._
import akka.remote.RemoteScope
import akka.util._

import java.util.concurrent.TimeUnit

import scala.concurrent._
import scala.concurrent.duration._

object ClientApp {
def main(args: Array[String]): Unit = {
val system = ActorSystem("LocalSystem", ConfigFactory.load("client"))

// get the remote actor via the server actor system's address
val serverAddress = AddressFromURIString("akka.tcp://ServerActorSystem@127.0.0.1:2552")
val actor = system.actorOf(Props[ServerActor].withDeploy(Deploy(scope = RemoteScope(serverAddress))))

// invoke the remote actor via a client actor.
// val remotePath = "akka.tcp://ServerActorSystem@127.0.0.1:2552/user/serverActor"
// val actor = system.actorOf(Props(classOf[ClientActor], remotePath), "clientActor")

buildReaper(system, actor)

// tell
actor ! 10000L

waitShutdown(system, actor)
}

private def buildReaper(system: ActorSystem, actor: ActorRef): Unit = {
import Reaper._
val reaper = system.actorOf(Props(classOf[ProductionReaper]))

// Watch the action
reaper ! WatchMe(actor)
}

private def waitShutdown(system: ActorSystem, actor: ActorRef): Unit = {
// trigger the shutdown operation in ProductionReaper
system.stop(actor)

// wait to shutdown
Await.result(system.whenTerminated, 60.seconds)
}
}

ProductionReaper.scala

class ProductionReaper extends Reaper {
// Shutdown
def allSoulsReaped(): Unit = {
context.system.terminate()
}
}

Reaper.scala

import akka.actor.{Actor, ActorRef, Terminated}
import scala.collection.mutable.ArrayBuffer

object Reaper {
// Used by others to register an Actor for watching
case class WatchMe(ref: ActorRef)
}

abstract class Reaper extends Actor {
import Reaper._

// Keep track of what we're watching
val watched = ArrayBuffer.empty[ActorRef]

// Derivations need to implement this method.  It's the
// hook that's called when everything's dead
def allSoulsReaped(): Unit

// Watch and check for termination
final def receive = {
case WatchMe(ref) =>
context.watch(ref)
watched += ref
case Terminated(ref) =>
watched -= ref
if (watched.isEmpty) allSoulsReaped()
}
}

ServerActor.scala

import akka.actor.Actor
import akka.actor.Props
import akka.event.Logging

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

class ServerActor extends Actor {
val log = Logging(context.system, this)

def receive = {
case n: Long =>
squareSum(n)
}

private def squareSum(n: Long): Long = {
val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)

val squareSum = sc.parallelize(1L until n).map { i =>
i * i
}.reduce(_ + _)

log.info(s"============== The square sum of $n is $squareSum. ==============")

squareSum
}
}

ServerApp.scala

import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import akka.actor.ActorSystem
import akka.actor.Props

object ServerApp {
def main(args: Array[String]): Unit = {
val system = ActorSystem("ServerActorSystem")
val actor = system.actorOf(Props[ServerActor], name = "serverActor")
}
}

构建工程

sbt package

测试应用

启动Spark服务

$SPARK_HOME/sbin/start-master.sh

7077

$SPARK_HOME/sbin/start-slave.sh spark://$(hostname):7077

启动Akka Server应用

$SPARK_HOME/bin/spark-submit --master spark://$(hostname):7077 --class ServerApp target/scala-2.11/akka-sample-app_2.11-1.0.jar

启动Akka Client应用

java -classpath ./target/scala-2.11/akka-sample-app_2.11-1.0.jar:$AKKA_HOME/lib/akka/*:$SCALA_HOME/lib/* ClientApp
# or
# $SPARK_HOME/bin/spark-submit --master spark://$(hostname):7077 --class ClientApp target/scala-2.11/akka-sample-app_2.11-1.0.jar

总结

下面请看

参照