文章目录

简介
- UML类图
- 创建过程
实现
- ShuffleClient接口定义
- 抽象类块传输服务BlockTransferService
- 实现类NettyBlockTransferService
- - 初始化
  - - RPC服务RpcServer
    - 传输上下文TransportContext
    - RPC客户端工厂TransportClientFactory
    - Netty服务器TransportServer
  - 获取远程shuffle文件
  - 上传shuffle文件到远程节点

shuffle服务与客户端简介实现

简介

ShuffleClient

不仅是将shuffle文件上传到其他Executor或者下载远程Executor文件到本地的客户端，也是提供可以被其他Executor访问的shuffle服务。

Spark任务是分布式计算的，每个Task运行在不同的机器上。map任务执行完成后会将输出结果存储到任务执行所在的机器的本地存储中，并通过本地的 MapOutputTrackerWorker 向 Driver 中 DAGScheduler 里的 MapOutputTrackerMaster 汇报。reduce任务很可能和map任务不在同一台机器上执行，reduce任务执行前首先会向本地 MapOutputTrackerWorker 请求map任务输出结果（如果没有则其向 Driver 中 DAGScheduler 里的 MapOutputTrackerMaster 获取）。得到地址后，reduce任务会通过 ShuffleClient 远程下载map任务的中间输出。因此， ShuffleClient 是Spark计算框架中的一个重要组件。

UML类图

ShuffleClient

的UML类图如下：

shuffle服务与客户端简介实现

创建过程

ShuffleClient

是在BlockManager中创建的，根据

spark.shuffle.service.enabled

配置来决定是否启用外部服务作为shuffle服务，默认使用构造函数中传入的

blockTransferService

：

// BlockManager.scala
private[spark] val externalShuffleServiceEnabled =
  conf.getBoolean("spark.shuffle.service.enabled", false)
// 读取其他executor的shuffle文件的客户端。它或者是一个外部服务，或者仅仅是标准的BlockTransferService
// 直接地连接到其他executors。
private[spark] val shuffleClient = if (externalShuffleServiceEnabled) {
  val transConf = SparkTransportConf.fromSparkConf(conf, "shuffle", numUsableCores)
  new ExternalShuffleClient(transConf, securityManager, securityManager.isAuthenticationEnabled())
} else {
  blockTransferService
}

BlockManager是在SparkEnv中创建的，并同时创建了

blockTransferService

服务并将其传入BlockManager的构造函数中。

blockTransferService

的实际实现类为

NettyBlockTransferService

：Spark和Hadoop一样，都采用Netty作为shuffle服务。下面我们以默认的

NettyBlockTransferService

为例进行介绍。

// SparkEnv.scala
val blockTransferService =
  new NettyBlockTransferService(conf, securityManager, bindAddress, advertiseAddress,
    blockManagerPort, numUsableCores)

val blockManagerMaster = new BlockManagerMaster(registerOrLookupEndpoint(
  BlockManagerMaster.DRIVER_ENDPOINT_NAME,
  new BlockManagerMasterEndpoint(rpcEnv, isLocal, conf, listenerBus)),
  conf, isDriver)

// NB: blockManager is not valid until initialize() is called later.
val blockManager = new BlockManager(executorId, rpcEnv, blockManagerMaster,
  serializerManager, conf, memoryManager, mapOutputTracker, shuffleManager, 
  blockTransferService, securityManager, numUsableCores

实现

ShuffleClient接口定义

首先看下ShuffleClient接口类的定义如下。ShuffleClient定义了从executor或者外部服务读取shuffle文件数据的接口。

/** 提供了一个从executor或者外部服务读取shuffle文件的接口。 */
public abstract class ShuffleClient implements Closeable {

  /**
   * 初始化ShuffleClient，指定当前executor的appId。
   * 必须在ShuffleClient其他方法调用前被调用。
   */
  public void init(String appId) { }

  /**
   * 从远程节点异步获取一组blocks
   *
   * 注意：该API接口接收数组参数，所以可以实现批量请求。另外，该方法没有返回一个future对象，所以子类
   * 实现可以在一个block获取成功后立即回调onBlockFetchSuccess，而不是等待所有的blocks都获取成功。
   */
  public abstract void fetchBlocks(
      String host,
      int port,
      String execId,
      String[] blockIds,
      BlockFetchingListener listener,
      TempShuffleFileManager tempShuffleFileManager);
}

抽象类块传输服务BlockTransferService

BlockTransferService

是继承自

ShuffleClient

接口的抽象类，负责数据的传输。其中定义了blocks的批量获取、单个获取和单个同步或异步上传的接口。类的定义如下：

private[spark]
abstract class BlockTransferService extends ShuffleClient with Closeable with Logging {
  // 通过提供可以用来获取和保存本地blocks块的BlockDataManager来初始化传输服务。
  def init(blockDataManager: BlockDataManager): Unit

  // 关闭服务
  def close(): Unit

  // 服务监听的端口号，只有在[[init]]调用后才可用。
  def port: Int

  // 服务监听的主机名，只有在[[init]]调用后才可用。
  def hostName: String

  /**
   * 从远程节点异步获取一组blocks
   *
   * 注意：该API接口接收数组接口，所以可以实现批量请求。另外，该方法没有返回一个future对象，所以子类
   * 实现可以在一个block获取成功后立即回调onBlockFetchSuccess，而不是等待所有的blocks都获取成功。
   */
  override def fetchBlocks(
      host: String,
      port: Int,
      execId: String,
      blockIds: Array[String],
      listener: BlockFetchingListener,
      shuffleFiles: Array[File]): Unit

  // 上传单个block块到远程节点，仅在[[init]]之后才可使用。
  def uploadBlock(
      hostname: String,
      port: Int,
      execId: String,
      blockId: BlockId,
      blockData: ManagedBuffer,
      level: StorageLevel,
      classTag: ClassTag[_]): Future[Unit]

  /**
    * 一个特殊的[[fetchBlocks]]的例子，它阻塞式地读取一个block块。
    * 只有在调用[[init]]后才可以使用它。
   */
  def fetchBlockSync(host: String, port: Int, execId: String, blockId: String): ManagedBuffer = {
    // 监控等待的线程.
    val result = Promise[ManagedBuffer]()
    fetchBlocks(host, port, execId, Array(blockId),
      new BlockFetchingListener {
        override def onBlockFetchFailure(blockId: String, exception: Throwable): Unit = {
          result.failure(exception)
        }
        override def onBlockFetchSuccess(blockId: String, data: ManagedBuffer): Unit = {
          val ret = ByteBuffer.allocate(data.size.toInt)
          ret.put(data.nioByteBuffer())
          ret.flip()
          result.success(new NioManagedBuffer(ret))
        }
      }, shuffleFiles = null)
    ThreadUtils.awaitResult(result.future, Duration.Inf)
  }

  /**
    * 上传单个block块到远程节点，仅在[[init]]调用之后才可用。
    * 该方法类似于[[uploadBlock]]方法，除了该方法会阻塞线程直到block上传完成。
   */
  def uploadBlockSync(
      hostname: String,
      port: Int,
      execId: String,
      blockId: BlockId,
      blockData: ManagedBuffer,
      level: StorageLevel,
      classTag: ClassTag[_]): Unit = {
    val future = uploadBlock(hostname, port, execId, blockId, blockData, level, classTag)
    ThreadUtils.awaitResult(future, Duration.Inf)
  }
}

实现类NettyBlockTransferService

BlockTransferService

的实现类实现为

NettyBlockTransferService

，它使用Netty异步时间驱动的网络应用框架，获取和上传远程节点上的Block集合。

初始化

NettyBlockTransferService的各项功能只能在init方法执行完成后才可用，初始化包含四大部分：

创建RPC服务RpcServer（实现子类NettyBlockRpcServer）；
构造传输上下文TransportContext；
创建RPC客户端工厂TransportClientFactory；

创建Netty服务器TransportServer；

// NettyBlockTransferService.scala
  override def init(blockDataManager: BlockDataManager): Unit = {
    // 1.创建RpcServer；
    val rpcHandler = new NettyBlockRpcServer(conf.getAppId, serializer, blockDataManager)
    var serverBootstrap: Option[TransportServerBootstrap] = None
    var clientBootstrap: Option[TransportClientBootstrap] = None
    if (authEnabled) {
      serverBootstrap = Some(new AuthServerBootstrap(transportConf, securityManager))
      clientBootstrap = Some(new AuthClientBootstrap(transportConf, conf.getAppId, securityManager))
    }
    // 2.构建TransportContext;
    transportContext = new TransportContext(transportConf, rpcHandler)
    clientFactory = transportContext.createClientFactory(clientBootstrap.toSeq.asJava)
    server = createServer(serverBootstrap.toList)
    appId = conf.getAppId
    logInfo(s"Server created on ${hostName}:${server.getPort}")
  }

RPC服务RpcServer

当map任务与reduce任务处于不同节点时，reduce任务需要从远端节点下载map任务的中间输出。NettyBlockRpcServer提供了下载Block文件的功能，并为了容错需要将Block的数据备份到其他节点上，还提供了上传Block文件的RPC服务。

/**
 * 服务通过为每个block请求简单地注册一个chunk来打开block块。用
 * 处理打开及上传BlockManager管理的任意blocks块。
 *
 * 打开的blocks被通过"一对一"的策略进行注册，意味着每个传输层的chunk对应一个Spark层的shuffle块。
 */
class NettyBlockRpcServer(
    appId: String,
    serializer: Serializer,
    blockManager: BlockDataManager)
  extends RpcHandler with Logging {...}

传输上下文TransportContext

TransportContext用于维护传输上下文，可以创建Netty服务和Netty访问的客户端。TransportContext的组成如下：

TransportConf：主要控制Netty框架提供的shuffle的I/O交互的客户端和服务端线程数量；
RpcHandler：负责shuffle的I/O服务端在接收到客户端的RPC请求后，提供打开Block或者上传Block的RPC处理，此处即为NettyBlockRpcServer；
decoder：在shuffle的I/O服务端对客户端传来的ByteBuf进行解析，防止丢包和解析错误；

encoder：在shuffle的I/O客户端对消息内容进行编码，防止服务端丢包和解析错误；

/**
   * 包含创建TransportServer和TransportClientFactory的上下文信息，
   * 并用`org.apache.spark.network.server.TransportChannelHandler`来设置Netty Channel管道。
   *
   * TransportClient提供了两个通信协议：控制层面的RPCs和数据层面的"chunk fetching"。
   * RPCs的处理是在TransportContext范围之外执行的（例如，通过一个用户提供的处理方法）。
   * 并且，它负责设置流，这些流可以利用零拷贝IO在chunks中来流式传输数据。
   *
   * TransportServer和TransportClientFactory为每个通道创建一个TransportChannelHandler。
   * 由于每个TransportChannelHandler都包含TransportClient，
   * 这使得服务器进程可以在现有channel通道上将消息发送回客户端。
   */
  public class TransportContext {
    private final TransportConf conf;
    private final RpcHandler rpcHandler;
  
    /**
     * 在类加载器切换为ExecutorClassLoader之前，强制创建MessageEncoder和MessageDecoder
     * (See SPARK-17714)
     */
    private static final MessageEncoder ENCODER = MessageEncoder.INSTANCE;
    private static final MessageDecoder DECODER = MessageDecoder.INSTANCE;
    /**
     * 初始化一个ClientFactory，它在返回新客户端之前运行给定的TransportClientBootstraps。
     * Bootstraps将同步执行，并且必须运行成功才能创建客户端。
     */
    public TransportClientFactory createClientFactory(List<TransportClientBootstrap> bootstraps) {
      return new TransportClientFactory(this, bootstraps);
    }
  
    public TransportClientFactory createClientFactory() {
      return createClientFactory(new ArrayList<>());
    }
  
    /** 创建尝试绑定到特定端口的服务器。 */
    public TransportServer createServer(int port, List<TransportServerBootstrap> bootstraps) {
      return new TransportServer(this, null, port, rpcHandler, bootstraps);
    }
    ...
  }

为什么需要MessageEncoder和MessageDecoder？因为在基于流的传输里（比如TCP/IP），接收到的数据首先会被存储到一个socket接收缓冲里。不幸的是，基于流的传输并不是一个数据包队列，而是一个字节队列。即使发送了2个独立的数据包，操作系统也不会作为2个消息处理，而仅仅认为是一连串的字节。因此不能保证远程写入的数据会被准确地解析。因此，接收方不管是客户端还是服务端，都应该把接收到的数据整理成一个或者多个更有意义并且让程序的逻辑更好理解的数据。

RPC客户端工厂TransportClientFactory

TransportClientFactory

由TransportContext的createClientFactory方法创建，是创建Netty客户端

TransportClient

的工厂类，用于向Netty服务端发送RPC请求。

TransportClientFactory

维护了一个client数组，当所需的客户端不存在的时候，创建一个新的网络连接，然后将连接保存到client数组中。其由以下部分组成：

clientBootstraps：用于缓存客户端列表；
connectionPool：用于缓存客户端连接；
numConnectionsPerPeer：节点之间取数据的连接数，可以使用属性 spark.shuffle.io.numConnectionsPerPeer 来配置，默认为1；
socketChannelClass：客户端channel被创建时使用的类，可以使用属性 spark.shuffle.io.mode 来配置，默认为NioSocketChannel；
workerGroup：根据Netty的规范，客户端只有work组，所以此处创建workerGroup，实际是NioEventLoopGroup；
pooledAllocator：汇集ByteBuf但对本地线程缓存禁用的分配器。

Netty服务器TransportServer

TransportServer由TransportContext的createServer方法创建，提供了Netty实现的服务器端，用于提供RPC服务（比如上传、下载等）。

获取远程shuffle文件

NettyBlockTransferService

作为shuffle客户端，其fetchBlocks方法可获取远程shuffle文件，实际是利用NettyBlockTransferService中创建的Netty服务。

NettyBlockTransferService

提供了非重试版本OneForOneBlockFetcher和重试版本RetryingBlockFetcher的BlockFetcher，重试次数通过参数

spark.[module].io.maxRetries

进行配置，默认是重试3次。

shuffle服务与客户端简介实现

override def fetchBlocks(

host: String,

port: Int,

execId: String,

blockIds: Array[String],

listener: BlockFetchingListener,

tempShuffleFileManager: TempShuffleFileManager): Unit = {

logTrace(s"Fetch blocks from h o s t : host: host:port (executor id $execId)")

try {

val blockFetchStarter = new RetryingBlockFetcher.BlockFetchStarter {

override def createAndStart(blockIds: Array[String], listener: BlockFetchingListener) {

// clientFactory维护了一个client数组，获取或者创建一个与目标主机和端口匹配的socket连接

val client = clientFactory.createClient(host, port)

new OneForOneBlockFetcher(client, appId, execId, blockIds, listener,

transportConf, tempShuffleFileManager).start()

}

val maxRetries = transportConf.maxIORetries()
    if (maxRetries > 0) {
      // 注意这个Fetcher会正确处理maxRetries == 0；我们判断避免它是以防代码中有bug。
	  // 一旦我们确定代码的稳定性，我们将会删除该if语句。
      new RetryingBlockFetcher(transportConf, blockFetchStarter, blockIds, listener).start()
    } else {
      blockFetchStarter.createAndStart(blockIds, listener)
    }
  } catch {
    case e: Exception =>
      logError("Exception while beginning fetchBlocks", e)
      blockIds.foreach(listener.onBlockFetchFailure(_, e))
  }
}

上传shuffle文件到远程节点

NettyBlockTransferService

本身也利用创建的Netty服务实现uploadBlock方法为其他远程Executor提供Block的读取服务，其实现流程如下：

创建Netty服务的客户端；
将Block的存储级别StorageLevel和ClassTag序列化；
将Block的字节数据ByteBuffer转化为数组，以便于序列化传输；
将由appId、execId、blockId、序列化的StorageLevel和ClassTag、以及转换为数组的Block封装的UploadBlock消息序列化为字节数组；
最终调用Netty客户端的sendRpc方法将字节数组上传，回调函数RpcResponseCallback根据RPC的结果更改上传状态。

shuffle服务与客户端简介实现

override def uploadBlock(

hostname: String,

port: Int,

execId: String,

blockId: BlockId,

blockData: ManagedBuffer,

level: StorageLevel,

classTag: ClassTag[_]): Future[Unit] = {

val result = PromiseUnit

val client = clientFactory.createClient(hostname, port)

// 使用JavaSerializer将StorageLevel和ClassTag序列化为字节。
	  // 其他所有内容都使用我们的二进制协议编码。
	  val metadata = JavaUtils.bufferToArray(serializer.newInstance().serialize((level, classTag)))
	
	  // 将nio缓冲区转换或复制到数组中以对其进行序列化。
	  val array = JavaUtils.bufferToArray(blockData.nioByteBuffer())
	  // 通过Client发送UploadBlock消息
	  client.sendRpc(new UploadBlock(appId, execId, blockId.toString, metadata, array).toByteBuffer,
	    new RpcResponseCallback {
	      override def onSuccess(response: ByteBuffer): Unit = {
	        logTrace(s"Successfully uploaded block $blockId")
	        result.success((): Unit)
	      }
	      override def onFailure(e: Throwable): Unit = {
	        logError(s"Error while uploading block $blockId", e)
	        result.failure(e)
	      }
	    })
	
	  result.future
	}

shuffle服务与客户端简介实现

文章目录

简介

UML类图

创建过程

实现

ShuffleClient接口定义

抽象类块传输服务BlockTransferService

实现类NettyBlockTransferService

初始化

RPC服务RpcServer

传输上下文TransportContext

RPC客户端工厂TransportClientFactory

Netty服务器TransportServer

获取远程shuffle文件

上传shuffle文件到远程节点

继续阅读

pyspark调用spark以及执行带in语句参数的hql示例

用写sql的思路写 pyspark

pyspark学习(一)—pyspark的安装与基础语法一 Pysaprk的安装二：pyspark的简单语法END

【Spark Mllib】K-均值聚类——电影类型K-均值聚类数据特征提取

一篇文章让你精通Java JSP规范

世界因大数据而改变

Spark的RDD转换算子-双value型Spark的RDD转换算子-双value型

SparkSQL项目练习1 准备数据2 需求：各区域热门商品Top3

延云行业搜索数据库在大数据生态中位置和重要性大数据的挑战大数据技术的现状延云行业搜索数据库

Spark在windows环境里跑时报错找不到org.apache.hadoop.fs.FSDataInputStream

Spark流式分析系统实现流式实时日志分析系统

Scala和Java二种方式实战Spark Streaming开发

Spark基础:Spark简介及特点,运行模式,安装Spark,Driver与Executor,Local模式,Standalone模式,Yarn模式,Mesos模式,WordCount案例,HA配置第1章 Spark概述第2章 Spark运行模式第3章案例实操

Spark实现wordcount

大数据排错SparkSpark集群启动时候，JAVA_HOME is not sethadoop集群，某台服务器jps无任何输出IDEAkafkahadoopspark sqlfile permissionsIDEA本地测试 - OutOfMemoryError: GC overhead limit exceededhdfs负载均衡

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结