理解Hadoop

忘是亡心i 2022-04-14 02:22 201阅读 0赞

# Hadoop Haloop #

hadoop解决迭代的MapReduce问题，有很多大数据分析需要迭代计算，而MapReduce框架对迭代计算支持度不够

### MapReduce产生的问题 ###

1.  每次执行都需要重新装在数据，重新处理，但迭代过程中，有动态数据和静态数据两类，静态数据处理带来额外的开销
2.  迭代终止稳定点的判断过程在每次迭代中需要额外的MapReduce计算

### MapReduce局限性 ###

1.  任务调度开销
2.  磁盘读取开销
3.  网络传输开销

### Hadoop解决思路 ###

1.  对于任务调度的开销：把循环体控制在job内，避免多次重复启动(Loop Control模块)
2.  对于磁盘读取开销：实现了MapReduce静态数据缓存(Task Scheduler促成数据本地化调度)
3.  对于网络传输开销：将终止条件的判断变得稳定和高效

### 寻找多跳邻居算法来理解Hadoop ###

#### 多跳邻居算法数学表示![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70][] ####

F表是邻居表，MR1是表的连接操作和表的投影操作，很容易理解  
MR2是过滤老朋友的操作，例如Eric的两跳邻居可以从Eric到Allice再回到Eric，因此需要把之前的重复朋友去掉，防止寻找的朋友回到了之前迭代的结果，因此需要减掉，防止在中间结果的地方一致迭代。

#### Hadoop代码形式main函数 ####

Main：
    Job job = new Job()
    job.AddMap(MapJoin, 1)
    job.AddReduce(ReduceJoin, 2) #MR1
    job.AddMap(MapDistinct, 1)
    job.AddReduce(ReduceDistinct, 2) #MR2
    job.setDistanceMeasure(ResultDistance)#由ResultDistance函数控制迭代结束
    job.setFixedPointThreshold(1) #ResultDistance函数的结果小于1的时候停止迭代
    job.setMaxNumOfIteration(2)#寻找2跳邻居，迭代两轮
    job.setInput(IterationInput)#每一轮迭代的输入
    job.setStepInput(StepInput)#一个MapReduce任务每一个步骤的输入
    job.AddInvariantTable(#1) #不变的表1
    job.SetReducerInputCache(true) #输入缓存
    job.summit()

#### MapJoin ####

使用的是Si.name2也就是values进行的连接操作，初始的时候，输出的是F表\#1，后期输出的是Si-1表\#2

Input:Key k, Value v, int interation
    if v from F
    	then output(v.name1, v.name2, #1)
    else
    	output(v.name2, v.name1, #2)#Si-1的value为name1，后面进行输出
    end if

#### ReduceJoin ####

做笛卡尔积运算  
输出同一个name1或者name2的姓名以及间接关联的邻居姓名

Input:Key key, set values, set invariantValues, int iteration
    output(product(values, invariantValues))

#### MapDistinct ####

输出为U Sj的两个姓名以及迭代下标

Input: Key key, value v, int iteration
    output(v.name1, v.name2, iteration)

#### ReduceDistinct ####

统计U Sj，也就是T3，并且用T2-T3

Input:Key key, set values, int iteration
    for name in values
    	if name.iteration < iteration
    		then set_odd.add(name)
    	else
    		set_new.add(name)
    end for
    output(product(key, distinct(set_new-set_odd)))

#### InterationInput ####

MR1的输入

Input:int iteration
    if iteration == 1 then
    	return F U ^S0
    else
    	return ^Si-1

#### StepInput ####

MR2的输入

Input:int step
    if step == 2
    	return U ^Sj

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]: /images/20220414/7f822746633c42e0a6976b442948e77f.png