Please enable JavaScript.

Coggle requires JavaScript to display documents.

Hadoop (Components (Mapper p72 (MapRunnable (more generic interface,…

- - - - HashPartitioner (default)
        
        take hash of the key, and modulo by number of reducers
      - TotalOrderPartitioner
        
        ensure all keys in a partition is smaller than ones in another
        
        :question:Sampling via IntervalSampler/RandomSampler/SplitSampler etc on client side to get split points
        
        TotalOrderPartitioner save the split points into a trie to fast locate partition of each given key, generating partitions of #reducers
        
        more balanced sampling == more efficiency
        
        usage
        
        TeraSort
        
        HBase batch import
  - - - more generic interface, custom handling k-v pairs
  - - - (logically) split input file into given #splits, each only has metadata like (file, start, length, hosts), not really splitting on disk
      - #splits determines #mappers for the task, :star:better make split size == block size
      - :question:host selection should maximize data locality.
        
        sort racks containing the InputSplit by size,
        
        sort nodes containing the InputSplit in each rack by size,
        
        select hosts of top N nodes (N = #replicas)
    - - iterator to get key-value pairs from InputSplit
      - determine boundary of records
      - decode key-value pairs
  - - - return RecordWriter which is used to write key-value pairs to file (in OutputCollector.collect(newK, newV))
    - - called before submitting job to JobTracker, to verify output validity (path exists? etc.)
    - - to optimize slow tasks, the same task is run in multiple places and take the first completed as result.
        the competing tasks all write to tmp file (handled by OutputCommitter)
      - when task is done, an empty file named _SUCCESS is created in the output dir
  - - - add EagerTaskInitializationListener and JobQueueJobInProgressListener to JobTracker
      - when job added/removed/updated, JobTracker invoke registered listeners
  - - - addCache[File|Archive]
      - addFileToClassPath
      - createSymlink
      - getLocalCache[Archives|Files]
    - - user must implement Tool interface in MapReduce program
  - - - startTracker - 创建JobTracker对象及初始化一些重要变量
        
        Access Management p142
        
        ACLsManager
        
        QueueManager
        
        JobACLsManager
        
        DNSToSwitchMapper
        
        implement to define conversion from DNS/IP to node location like /dc2/rack1/node4
      - offerService
        
        expireTrackersThread
        
        发现和清理死掉的TaskTracker（心跳超时），从数据结构trackerToJobsToCleanup，trackerToTasksToCleanup，trackerToTaskMap，trackerToMarkedTasksMap中清除之，标记正在运行的task为KILLED_UNCLEAN
        
        retireJobsThread
        
        清理长时间驻留内存的已完成task信息
        
        expireLaunchingTaskThread
        
        发现已被分配给某个TaskTracker但一直未回报信息的task，标记为FAILED
        
        completedJobsStoreThread
        
        将完成的job信息保存到HDFS
    - - JobHistory类记录关键时间日志
      - RecoveryManager类通过日志恢复job运行状态
  - - - pull模式，JobTracker从不主动联系，TaskTracker定期发送心跳通信
      - JobTracker根据集群信息动态调整心跳间隔，将下一次心跳间隔回复给TaskTracker
      - task运行失败时TaskTracker缩短心跳间隔，以尽快重试任务
      - 步骤
        
        检查JobTracker／TaskTracker版本是否匹配
        
        检查磁盘是否损坏，是则重新初始化TT
        
        发送心跳，接收回复并执行JT传达的命令
    - - LaunchTaskAction p216
        
        TT出现空闲资源后，心跳请求新任务，JT通过TaskScheduler安排任务并返回，TT创建JVM启动任务
        
        分为计算型任务（Map／Reduce，由专门scheduler调度）和辅助型任务（setup／cleanup，由JobTracker调度，优先级高于计算型任务）
      - CommitTaskAction p206
        
        计算结果先写入临时目录并通过心跳向JT发出提交请求，JT批准后才可以将结果转移到最终目录
        
        2PC (pull-mode)
        
        TaskAttempt完成计算后运行状态RUNNING -> COMMIT_PENDING，RPC告知TT
        
        TT得知一个TA为COMMIT_PENDING状态后，立刻缩短心跳间隔以快速汇报给JT
        
        JT检查如果该TA是该TaskInProgress第一个完成的，回复CommitTaskAction批准提交
        
        TT将该TA加入commitResponses列表
        
        TA通过RPC定期检查自己是否在commitResponses列表内，是则转移结果到最终目录并确认提交完成给TT
        
        TT将该TA状态改为SUCCEEDED，下次心跳通知JT
        
        pull模式，TA定期查询TT，以分散TT的工作负载
      - KillTaskAction p208
        
        JobClient向JT发送KillTask请求，JT将该TA加入tasksToKill列表
        
        TA所在TT心跳时，JT回复KillTaskAction及相关信息
        
        TT将该TA移除runningJobs，状态改为KILLED_UNCLEAN，通知directoryCleanupThread线程清理其工作目录，释放所占slot，缩短心跳间隔
        
        JT命令TT启动一个cleanUpTask来清理TA已经写入HDFS的数据，TT收到后启动JVM执行任务，完成后通知JT更新TA状态为KILLED
      - KillJobAction p210
        
        JT通过心跳向该job相关的所有TT广播KillJobAction以清理空间
      - ReinitTrackerAction p212
        
        TT出现磁盘故障或JT回复ReinitTrackerAction时重新初始化自己，过程与启动一致
        
        TaskTracker状态不一致（JobTracker未刚刚重启，且TaskTracker并非初次连接，但最近的HeartbeatResponse丢失了）时，命令TaskTracker重新初始化
    - - TT配置多个挂载在不同磁盘的目录作为中间结果存放目录，通过轮询依次使用
- - - - define dependant job, Hadoop will run jobs in topological order
    - - WAITING
        
        dependant job not done yet
      - READY
        
        no pending dependancy, ready to run
      - RUNNING
        
        job executing, may end up in SUCCESS or FAILED
  - - - collect() will call map on the next Mapper, or write to file on last Mapper
  - - - Setup Task (Optional)
        
        update state to setup, call OutputCommitter.setupJob()
        
        when finished, job state change PREP -> RUNNING, start running Map task
      - Map Task
        
        execute Mappers, number is determined by input splits count
      - Reduce Task
        
        count configed by mapred.reduce.tasks (default 1)
        
        Hadoop starts Reduce tasks only after mapred.reduce.slowstart.completed.maps (default 5%) Map tasks are done
      - Cleanup Task (Optional)
        
        delete temporary files/dirs, set state to RUNNING -> SUCCEEDED once done
  - - - 推测出拖后腿的任务，为其启动一个并行的备份任务，取最先完成的为结果并取消另一个
      - 满足任一条件就启动备份任务
        
        该task尚未进入skip mode（不会同时启用备份任务和skip mode，都会减慢任务执行速度）
        
        该task没有其他正在运行的备份Task Attempt
        
        该task已经运行超过60s且落后于该job所有Task Attempts平均进度的20%
      - Longest Approximate Time to End p176
        
        基于task运行速度和task最大剩余时间
        
        根据历史数据识别快慢节点，将备份任务分配给快节点
        
        判断能否在TaskTracker X上为job J的某个task T启动备份任务
        
        progressRate
        
        单位时间内task进度的增长率
        
        Job J所有task进度增长率的标准方差
        
        X必须是一个快节点，通过X运行该job其他任务时和集群全局的性能表现评估
        
        检查job J已经启动的task数是否超过限制
        
        找出可能拖后腿的candidate任务
        
        按照运行剩余时间从大到小排序，为最大的task启动备份
        
        :!:静态方式计算任务进度，性能低下，进度估算不准确导致误判并启动不必要的备份，未针对任务类型（Map／Reduce）更细致地优化
        SAMR是改进版
      - MapReduce2.0 算法关注备份任务是否有潜力超过原任务
  - - - 通过RecordReader读取InputSplit并解析成key-value pairs
        
        调用用户定义的map()进行处理产生新的k-v pairs
        
        调用OutputCollector.collect，将结果用Partitioner分片，写入环形内存缓冲区MapOutputBuffer
        
        当内存缓冲区满，排序（多轴快速排序 p229 ），压缩缓存中数据并写入磁盘临时文件
        
        (Optional) 调用Combiner以分片为单位合并计算结果
        
        所有数据处理完成后，Merger合并（基于小顶堆的多轮归并排序 p231 ）所有文件成只有一个文件
      - Collect p239
        
        map方法处理完一个键值对后调用OutputCollector.collect，调用Partitioner.getPartition获取键值对所属分区号，将三元组<key, value, partition>传给MapOutputBuffer.collect处理
        
        run() -> runOldMapper() -> 创建OldOutputCollector -> 创建MapRunnable迭代调用map
        
        MapOutputBuffer p238
        
        内部采用两级索引结构，包含三个环形内存缓冲区
        
        kvoffsets
        
        偏移量索引，保存键值对在位置索引kvindices中的偏移量
        
        kvindices
        
        保存键值对在数据缓冲区kvbuffer内的起始位置
        
        kvbuffer
        
        环形数据缓冲区，保存实际数据，用量超过io.sort.spill.percent后触发SpillThread将数据写入磁盘
        
        collect／write方法和spillThread通过可重入互斥锁spillLock和其上条件变量spillDone，spillReady进行同步
        
        Hadoop0.21采用共享环形缓冲区，无须再设置io.sort.record.percent。用指针equator界定索引和数据的共同起始点，各自朝相反方向扩张
      - Spill p247
        
        对内存缓冲区数据进行快速排序，先按分片再按key大小排序，Optional调用Combiner合并数据，写入磁盘文件
      - Combine p247
        
        基于最小堆的多轮递归合并，每次合并前100（默认）个文件直到最后只有一个文件，避免同时打开大量文件的开销
    - - Shuffle／Copy - 从各个Map task远程读取一片数据，存入内存或溢写磁盘（ReduceCopier类执行）
        
        Merge - 远程读取过程中开启两个后台线程对内存和磁盘数据进行合并（ReduceCopier类执行）
        
        Sort - 对各个Map task传来的有序数据片进行归并排序
        
        Reduce - 调用用户定义的reduce方法处理数据
        
        Write - 将计算结果写入HDFS
      - Sort 和 Reduce并行进行。Sort阶段Reduce Task为内存和磁盘文件建立小顶堆，维护指向堆顶的迭代器。
        Reduce task不断移动迭代器，将key相同的数据顺次交给reduce方法处理
- - - - in-mapper combining
        
        use an internal data structure to accumulate state during mapper execution
        
        :!: may cause OOM if size is too large
        
        more efficient and controllable than combiner
    - - in-reducer sorting can bottleneck on memory
      - value-to-key conversion
        
        move the value you want to sort on into a composite key, and
        define a custom partitioner to properly group the composite keys to the same reducer
    - - reduce-side join
        
        map over both datasets and emit the join key as intermediate key
        
        one-to-many
        
        create composite key of (join key, row id), define sort order to first sort by join key,
        then sort all rows from dataset1 before all rows from dataset2
        
        many-to-many
        
        for each join key, reducer buffers all rows from the smaller dataset,
        then cross-product with rows from the other dataset
        
        parallel sort-merge join
      - map-side join
        
        align both datasets by the join key in input split for each Mapper, and perform join within mapper
        
        far more efficient than reduce-side join, because no need to shuffle the datasets over network
        
        :!: the reducers generating data for later map-side join MUST NOT emit any key but the one they are currently processing
        
        merge join
      - memory-backed join
        
        simple hash join
        
        load the smaller dataset into memory, and map over the larger dataset to join with the in-memory one
        
        divide the smaller dataset into shards if not fit into memory, or use external distributed key/value store to hold dataset
- - - - assignTasks
        
        从JobTracker获取集群运行时信息，以优化分配task给TaskTracker
      - TaskScheduler通过taskTrackerManager.addJobInProgressListener注册回调以响应事件
    - - FIFO (JobQueueTaskScheduler) p186
        
        EagerTaskInitializationListener
        
        初始化提交的作业，优先选择高优先级的，再选择提交事件更早的
        
        JobbQueueJobInProgressListener
        
        维护作业调度顺序，优先选择高优先级的，再选择提交事件更早的
        
        assignTasks
        
        计算该TaskTracker可用slot数量，尽量均衡分布
        
        按调度顺序遍历job队列，调用JobInProgress的obtainNew[Map|Reduce]Task选择availableSlot个任务
  - - - 读取/proc下meminfo，cpuinfo，stat文件获取节点资源信息
- - - - 读写数据的最小单位，默认64MB
      - 更大的块可减少寻址开销
    - - 维护文件和索引结构
    - - 保存数据块
  - - - DistributedFileSystem.open() 打开文件系统连接
        
        DistributedFileSystem 访问NameNode获得文件块所在DataNode的地址列表
        
        通过FSDataInputStream依次从其中最近的DataNode读取数据
        
        关闭文件系统连接
    - - DistributedFileSystem.open() 打开文件系统连接
        
        DistributedFileSystem请求NameNode为即将写入的数据分配存储空间及相关信息
        
        副本地址一般选择一个不同节点，一个不同机架，一个不同DC
        
        提供FSData向NameNode分配的地址写入数据。数据会串流至所有副本地址
        
        所有副本确认写入完成后关闭DistributedFileSystem连接
      - 当前正在写入的块是对其他客户端不可见的
    - - 从HDFS并行复制大量数据
- - - - TaskTracker第一次汇报心跳后，JobTracker将其放入过期队列trackerExpiryQueue，并加入网络拓扑
      - TaskTracker之后每次心跳，JobTracker记录最近一次心跳时间TaskTrackerStatus.lastSeen
      - 线程expireTrackersThread周期性扫描过期队列trackerExpiryQueue，移除超时未心跳的TaskTracker
      - 已完成的Reduce和Map-only的Map任务无须重新运行，因为结果已经写入HDFS。
        其余情况必须重试，因为其他TaskTracker无法访问死掉TaskTracker上的本地数据
    - - 启发式算法推断出异常的TaskTracker进入灰名单，不再能接受新任务，直到一段时间后才能重获机会
      - 用户设定的脚本监控的异常TaskTracker进入黑名单，不再能接受新任务，不会再复活，直到监控脚本发现它又活过来了
      - job运行时动态生成黑名单。记录每个TaskTracker失败的Task Attempt数量，超过限制（mapred.max.tracker.failures）时加入黑名单
      - JobTracker维护一个环形桶数据结构，保存近期内一个TaskTracker对应的加入黑名单的次数
      - :star:核心思想 - 如果一个TaskTracker不健康，停止给它新任务，直到恢复健康
- - - - HadoopRPC采用反射，动态代理 p92
    - - TCP／IP Socket
    - - Netty，Reactor模式的event-driven IO模型