对Spark的那些【魔改】

比如在PSExecutorBackend 实现如下代码：

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = { 
    case Message.TensorFlowModelClean(modelPath) => { 
      logInfo("clean tensorflow model") 
      TFModelLoader.close(modelPath) 
      context.reply(true) 
    } 
    case Message.CopyModelToLocal(modelPath, destPath) => { 
      logInfo(s"copying model: ${modelPath} -> ${destPath}") 
      HDFSOperator.copyToLocalFile(destPath, modelPath, true) 
      context.reply(true) 
    } 
  } 

接着你就可以在Spark里写如下的代码调用了：

val psDriverBackend = runtime.asInstanceOf[SparkRuntime].psDriverBackend psDriverBackend.psDriverRpcEndpointRef.send(Message.TensorFlowModelClean("/tmp/ok")) 

是不是很酷。

修改闭包的序列化方式

Spark的任务调度开销非常大。对于一个复杂的任务，业务逻辑代码执行时间大约是3-7ms,但是整个spark运行的开销大概是1.3s左右。

经过详细dig发现，sparkContext里RDD转化时，会对函数进行clean操作，clean操作的过程中，默认会检查是不是能序列化(就是序列化一遍，没抛出异常就算可以序列化)。而序列化成本相当高(默认使用的JavaSerializer并且对于函数和任务序列化，是不可更改的)，单次序列化耗时就达到200ms左右，在local模式下对其进行优化，可以减少600ms左右的请求时间。

当然，需要申明的是，这个是针对local模式进行修改的。那具体怎么做的呢?

（编辑：核心网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!