通過spark-submit提交的任務都需要指定Main類作為程序的入口屿良，Main類執(zhí)行結束即Spark任務終結偎箫。如果需要通過外部程序實時向Spark任務提交數(shù)據并獲取結果又該如何呢窒升？
思路很簡單蓝晒，讓Spark任務的Main方法不終止声功，外部程序與Spark任務進行通信，交互數(shù)據仔引。
通信方式很多，比如Socket褐奥，netty或者內置Tomcat,Jetty等咖耘，不過考慮編碼的快捷，通過Akka是比較不錯的選擇撬码。
開發(fā)分為2部分儿倒。1.編寫Spark任務，該部分會提交到Spark集群中蚕泽。2.外部調用代碼吊骤，該部分模擬客戶端代碼丽惭。2者食用Akka Actor進行通信行您。
先看Spark任務部分
SparkConfig 定義SparkContext對象
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
object SparkConfig {
val conf = new SparkConf().setAppName("testSpark")
val sc = new SparkContext(conf)
}
DataService 作為調用Spark RDD操作的業(yè)務類瞧挤。
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import com.sam.spark.demo.data.config.SparkConfig
import scala.collection.mutable.ArrayBuffer
class DataService {
def handler(list: ArrayBuffer[String]) : String = {
val array = SparkConfig.sc.parallelize(list).max()
array
}
}
Worker Akka的Actor對象龄砰，接收外部入參肛宋，調用DataService對象恋捆，并返回結果
import akka.actor.Actor
import org.slf4j.LoggerFactory
import com.sam.spark.demo.data.service.DataService
import com.sam.spark.demo.akka.msg.TextMessage
import java.util.UUID
import scala.collection.mutable.ArrayBuffer
class Worker extends Actor {
val dataService = new DataService()
def receive = {
case x: ArrayBuffer[String] => {
val tm = new TextMessage()
tm.msg = dataService.handler(x)
sender ! tm
}
}
}
TextMessage 作為返回的消息對象
class TextMessage extends Serializable {
var msg : String = null
}
AkkaConfig Actor配置類,創(chuàng)建Worker對象
import akka.actor.ActorSystem
import akka.actor.Props
import com.typesafe.config.ConfigFactory
object AkkaConfig {
val system = ActorSystem("ReactiveEnterprise",ConfigFactory.load().getConfig("serverSystem"))
val workerRef = system.actorOf(Props[Worker], "worker")
}
程序入口類
import com.sam.spark.demo.akka.AkkaConfig
import com.sam.spark.demo.data.config.SparkConfig
import scala.concurrent.duration.Duration
import scala.concurrent.Await
import java.util.concurrent.TimeUnit
object AppStart {
def main(args: Array[String]): Unit = {
SparkConfig
AkkaConfig
}
}
pom.xml
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.6.3</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.10.5</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.6.0</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>Main類名</mainClass>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
</transformers>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
Akka配置文件
serverSystem {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
default-dispatcher {
throughput = 2
}
serializers {
java = "akka.serialization.JavaSerializer"
}
serialization-bindings {
"需要序列化的消息類名" = java
}
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "Akka Remote服務地址"
port = Akka Remote端口
}
}
}
}
打包
mvn clean scala:compile package -DskipTests=true
發(fā)布到Spark集群
./spark-1.6.3-bin-hadoop2.6/bin/spark-submit --class Main類名 --master spark://Spark Master地址 ./spark.demo-0.0.1-SNAPSHOT.jar
再看客戶端實現(xiàn)
本地Actor 獲取遠程ActorRef并發(fā)送消息
import akka.actor.Actor
import akka.actor.ActorSelection
import com.sam.spark.demo.akka.msg.TextMessage
import scala.collection.mutable.ArrayBuffer
class Client extends Actor {
var remoteActor : ActorSelection = context.actorSelection("akka.tcp://ReactiveEnterprise@10.16.64.146:2555/user/processManagers/worker")
override def receive: Receive = {
case msg: ArrayBuffer[String] => {
remoteActor ! msg
}
case msg: TextMessage => {
println(msg.msg)
}
}
}
本地Main類 模擬向遠端Actor發(fā)送消息
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props
import akka.pattern.Patterns
import scala.concurrent.duration.Duration
import scala.concurrent.Await
import akka.util.Timeout
import java.util.concurrent.TimeUnit
import java.util.UUID
import scala.collection.mutable.ArrayBuffer
object ClientStart {
def main(args: Array[String]): Unit = {
val serverSystem = ActorSystem("clientSystem", ConfigFactory.load().getConfig("clientSystem"))
val clientRef = serverSystem.actorOf(Props[Client], "client")
while (true) {
var list = new ArrayBuffer[String]
for (i <- 1 to 100) {
list += UUID.randomUUID().toString()
}
clientRef ! list
Thread.sleep(500)
}
// val future = Patterns.ask(clientRef, "world", Timeout.apply(10L, TimeUnit.SECONDS));
// val result = Await.result(future, Duration.create(10, TimeUnit.SECONDS));
// println(result)
}
}
Akka配置文件
clientSystem {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
default-dispatcher {
throughput = 2
}
serializers {
java = "akka.serialization.JavaSerializer"
}
serialization-bindings {
"需要序列化的消息類名" = java
}
}
}
}
