我们都知道Spark的每个task运行在不同的服务器节点上,map输出的结果直接存储到map任务所在服务器的存储体系中,reduce任务有可能不在同一台机器上运行,所以需要远程将多个map任务的中间结果fetch过来。那么我们就来学习下shuffleClient。shuffleClient存在于每个exeuctor的BlockManager中,它不光是将shuffle文件上传到其他executor或者下载到本地的客户端,也提供了可以被其他exeuctor访问的shuffle服务.当有外部的(其他节点)shuffleClient时,新建ExternalShuffleClient,默认为BlockTransferService.那么真正init的实现方法在NettyBlockTransferService中。

  

  如代码中所示,抽象类blockTransferservice继承自shuffleClientNettyBlockTransferService实现了shuffleClient的init抽象方法(竟然是java写的)进行初始化提供服务。初始化的过程为:创建NettyBlockRpcServer,构造TransportContext上下文,同时创建了clientFactory,最终创建了Netty服务器TransportServer,可修改属性spark.blockManager.port改变TransportServer的端口。

  我们会有疑问,上面那一坨,是干嘛的?我们都知道,map和reduce任务处于不同节点时,reduce任务需要从远端fetch map任务的中间结果输出,NettyBlockRpcServer提供打开,下载Block文件的功能(中间结果在backet中)。NettyBlockRpcServer为了容错,还会将数据备份到其他节点。在new 了之后会根据接收到的message消息,匹配是打开block还是上传block进行容错。如图:

  

  在new完NettyBlockRpcServer后,开始构造传输的上下文TransportContext.构造它的主要作用是,它将既可以创建Netty服务,也可以创建Netty访问客户端,主要包含:

  1、TransportConf,控制Netty框架提供的shuffle I/O交互的客户端和服务端线程数量(又发现新的参数)。

  2、RpcHandler,负责shuffle的I/O服务端在接受到客户端的RPC请求后,提供打开Block或者上传Block的RPC处理,就是刚才new的NettyBlockRpcServer,可以看到receive。

  3、decoder,在shuffle的I/O服务端对客户端传来的ByteBuf进行解析,防止丢包和解析错误

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

  

  那么为什么需要decoder、encoder呢,这里要补习下传输原理,一般基于TCP/IP的流传输中,接收到的数据首先会被存储到一个socket缓冲区中,基于流的传输并不是一个数据包的队列,而是一个字节队列。即使发送两个独立的数据包,操作系统也不会作为2个消息处理,而作为一连串的字节。也就是说 发送的数据可能是 ABC UID GDI ,应用程序读取的时候数据很可能被分成了 AB CUID G DI,所以应该把接收到的数据整理成一个或多个有意义能让程序的逻辑更好理解的数据。

  接下来,开始创建RPC客户端工程ClientFactory,它主要:1、缓存客户端列表。2、缓存客户端连接。3、节点之间取数据的连接数,通过spark.shuffle.io.numConnectionsPerPeer来配置,默认为1。4、客户端channel被创建时使用的类,可以使用属性spark.shuffle.io.mode来配置,默认为NioSocketChannel.(NIO还没仔细学习过,它的特点为所有的原始类型提供(Buffer)缓存支持,字符集编码解决方案,提供一个新的原始的I/O抽象Channel,支持锁和内存映射文件的文件访问接口;提供多路非阻塞的高伸缩性网络I/O)

  最终,createServer,看不懂NIO,回头恶补下。。

  

  那么下来,到了最重要的环节,获取远程shuffle文件,也就是fetch数据的过程。这个过程就是之前上面NettyBlockTransferService中的fetchBlocks方法(在shuffle过程中,可以通过container日志查看到fetch数据):

  

  可以从传入的参数中看到,会传入拉取节点的IP与PORT以及blockId信息,进行数据的拉取。

  那么之前,我们提到的上传shuffle文件,以便之前的拉取,也是先创建了Netty服务的客户端,同时我们可以看到它进行了serializer序列化并转化为了array()数组。随之将blockId、appId、execId等一起封装,调用Netty客户端的sendRpc方法将字节数组上传,同时毁掉函数RpcResponse-CallBack根据RPC的结果更改了上传状态。如下代码:

  今天到此为止,开始敲代码~

 

                  

 

Spark数据传输及ShuffleClient(源码阅读五)的更多相关文章

  1. Spark常用函数(源码阅读六)

    源码层面整理下我们常用的操作RDD数据处理与分析的函数,从而能更好的应用于工作中. 连接Hbase,读取hbase的过程,首先代码如下: def tableInitByTime(sc : SparkC ...

  2. JDK源码阅读(五)java.io.Serializable接口

    package java.io; public interface Serializable { } (1)实现Serializable接口的类,将会被提示提供一个 serialVersionUID ...

  3. Struts2源码阅读(一)_Struts2框架流程概述

    1. Struts2架构图  当外部的httpservletrequest到来时 ,初始到了servlet容器(所以虽然Servlet和Action是解耦合的,但是Action依旧能够通过httpse ...

  4. Spark源码阅读之存储体系--存储体系概述与shuffle服务

    一.概述 根据<深入理解Spark:核心思想与源码分析>一书,结合最新的spark源代码master分支进行源码阅读,对新版本的代码加上自己的一些理解,如有错误,希望指出. 1.块管理器B ...

  5. 【原】AFNetworking源码阅读(五)

    [原]AFNetworking源码阅读(五) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 上一篇中提及到了Multipart Request的构建方法- [AFHTTP ...

  6. 【原】SDWebImage源码阅读(五)

    [原]SDWebImage源码阅读(五) 本文转载请注明出处 —— polobymulberry-博客园 1. 前言 前面的代码并没有特意去讲SDWebImage的缓存机制,主要是想单独开一章节专门讲 ...

  7. 37 网络相关函数(五)——live555源码阅读(四)网络

    37 网络相关函数(五)——live555源码阅读(四)网络 37 网络相关函数(五)——live555源码阅读(四)网络 简介 10)MAKE_SOCKADDR_IN构建sockaddr_in结构体 ...

  8. Redis源码阅读(五)集群-故障迁移(上)

    Redis源码阅读(五)集群-故障迁移(上) 故障迁移是集群非常重要的功能:直白的说就是在集群中部分节点失效时,能将失效节点负责的键值对迁移到其他节点上,从而保证整个集群系统在部分节点失效后没有丢失数 ...

  9. 编译spark源码及塔建源码阅读环境

    编译spark源码及塔建源码阅读环境 (一),编译spark源码 1,更换maven的下载镜像: <mirrors> <!-- 阿里云仓库 --> <mirror> ...

随机推荐

  1. MySqlHelper、CacheHelper

    MySqlHelper代码: using System; using System.Collections; using System.Collections.Generic; using Syste ...

  2. 找到一款不错的网站压力测试工具webbench

    webbench最多可以模拟3万个并发连接去测试网站的负载能力,个人感觉要比Apache自带的ab压力测试工具好,安装使用也特别方便. 1.适用系统:Linux 2.编译安装: 引用 wget htt ...

  3. dwz中权限的控制

    很多人不明白用dwz要如何在没有登录的时候跳转到登录页面,没有权限的时候弹出提示. 其实,作者在设计的时候,已经完全考虑到了这些需求. 不管是navTab还是dialog,dwz的页面加载最终都是通过 ...

  4. Java-NIO-Selector

    扩展阅读: Java NIO类库Selector机制解析(上) Java NIO类库Selector机制解析(下) Java NIO的选择器 三个重要的类: 1,Selector 选择器,完成主要的选 ...

  5. oracle用户管理入门

    1.创建一个新用户(普通) create user 用户名 identified by 密码 示例,创建一个名叫小明的用户,密码为xm 2.以管理员的身份给普通用户修改密码(当普通用户忘掉自己的密码时 ...

  6. ssh相关问题

    问题1 一般错误信息为:ssh: connect to host localhost port 22: Connection refused 这种错误很主要的一个原因是sshd服务没有启动,先启动ss ...

  7. 如何利用百度orc实现验证码自动识别

    在爬取网站的时候都遇到过验证码,那么我们有什么方法让程序自动的识别验证码呢?其实网上已有很多打码平台,但是这些都是需要money.但对于仅仅爬取点数据而接入打码平台实属浪费.所以百度免费orc正好可以 ...

  8. Django App(一) StartApp

    经过配置Pycharm在上一次的笔记中,已经解决了编写Django web程序调试的问题,这篇将记录Django官网提供的例子程序!          1.查看Pycharm terminal是否可用 ...

  9. 转载,matla滤波函数

    转载地址http://blog.sina.com.cn/s/blog_6163bdeb0102e1dj.html 滤波器设计是一个创建满足指定滤波要求的滤波器参数的过程.滤波器的实现包括滤波器结构的选 ...

  10. [转][c++][跨平台]c++跨平台开发小结

    转自:https://blog.csdn.net/dj0379/article/details/53577135 linux编程与windows编程的差异之处: 1. 文件与目录的大小写以及路径分隔符 ...