我一直试图spark-shell在YARN client模式下运行,但是我遇到了很多ClosedChannelException错误.我正在使用spark 2.0.0 build for Hadoop 2.6.
以下是例外情况:
$ spark-2.0.0-bin-hadoop2.6/bin/spark-shell --master yarn --deploy-mode client
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel).
16/09/13 14:12:36 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
16/09/13 14:12:38 WARN yarn.Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
16/09/13 14:12:55 ERROR cluster.YarnClientSchedulerBackend: Yarn application has already exited with state FINISHED!
16/09/13 14:12:55 ERROR …问题陈述:我们将MySQL DB表中特定模块的用户的所有传入请求参数作为一行添加(这是一个巨大的数据).现在,我们想要设计一个进程,该进程将从该表中读取每条记录,并通过调用第三方API获得有关该用户请求的更多信息,之后它将把这个返回的元信息放在另一个表中.
目前的尝试:
我正在使用Scala + Slick来做到这一点.由于要读取的数据很大,我想一次一行地读取该表并进行处理.我尝试使用slick + akka流,但是我得到'java.util.concurrent.RejectedExecutionException'
以下是我尝试过的粗略逻辑,
implicit val system = ActorSystem("Example")
import system.dispatcher
implicit val materializer = ActorMaterializer()
val future = db.stream(SomeQuery.result)
Source.fromPublisher(future).map(row => {
id = dataEnrichmentAPI.process(row)
}).runForeach(id => println("Processed row : "+ id))
dataEnrichmentAPI.process:此函数进行第三方REST调用,并进行一些数据库查询以获取所需数据.这个数据库查询是使用'db.run'方法完成的,它也会一直等到它完成(使用Await)
例如,
def process(row: RequestRecord): Int = {
// SomeQuery2 = Check if data is already there in DB
val retId: Seq[Int] = Await.result(db.run(SomeQuery2.result), Duration.Inf)
if(retId.isEmpty){
val metaData = RestCall()
// SomeQuery3 = Store this metaData in DB
Await.result(db.run(SomeQuery3.result), …