对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模式举办修改的。那详细怎么做的呢?

（编辑：湖南网）

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