Hadoop-MapReduce的Shuffle机制

た入场券 2022-10-13 12:55 120阅读 0赞

## Shuffle机制 ##

### Shuffle机制 ###

Map 方法之后,Reduce 方法之前的数据处理过程称之为 Shuffle。即洗牌。其中涉及到分区、排序（快排）、Combiner、归并排序、压缩等数据处理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

### Partition 分区 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

最终要输出多少文件，就要分多少个区，默认的分区方法是hash分区

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]

其中的Partitioner的输入和输出对应的是map的输入和输出类型。

分区实操：

目的：将统计结果按不同的姓氏输出到不同的文件中；

输入数据如下：

1 caocao 123456 100 200
    2 liubei 002396 15 75
    3 sunquan 123456 200 150
    4 liuxian 002396 850 850
    5 liubei 200 300
    6 caozhi 200 300
    7 lvbu 200 300
    8 caopi 200 300
    9 sunce 200 300
    10 lvbu 200 300
    11 caopi 200 300
    12 sunce 200 300
    13 caocao 123456 123 123
    14 caocao 123456 223 44
    15 caocao 123456 434 55
    16 caocao 123456 567 66

期望输出数据：

姓氏为cao、liu、sund开头的用户分别放到独立的文件中，其余的放到一个文件中。

在**序列化实操**部分的代码上进行改进，

只需要修改驱动类，加入一下数据：

// 指定自定义数据分区
    job.setPartitionerClass(NamePartitioner.class);
    // 同时指定相应数量的reduceTask
    // ReduceTask默认是1，如果这里不设置分区数的话，会将执行的结果全部输出到一个文件中
    // 如果设置的值不是1且小于正确的分区数的话，那么将会报IO异常的错误
    // 如果设置的值不是1且大于正确的分区数的话，任务将会正常执行，只是多余的文件内容为空
    job.setNumReduceTasks(4);

并增加分区类NamePartitioner

public class NamePartitioner extends Partitioner<Text, UserFlowBean> { 
    
        @Override
        public int getPartition (Text text, UserFlowBean userFlowBean, int i) { 
    
            // key是用户名
            // value是登录登出信息
    
            // partition必须从0开始
            int partition = 3;
    
            String key = text.toString();
            if (key.startsWith("cao")) { 
                partition = 0;
            } else if (key.startsWith("liu")) { 
                partition = 1;
            } else if (key.startsWith("sun")) { 
                partition = 2;
            }
    
            return partition;
        }
    }

运行结果如下：

![在这里插入图片描述][20210707113652272.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]

### 排序 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]

排序的分类：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]

#### 自定义排序WritableComparable ####

原理分析：bean 对象做为 key 传输,需要实现 WritableComparable 接口重写compareTo 方法,就可以实现排序。

实操（全排序）：

根据**序列化实操**的案例进行修改，对总次数进行倒序排序。

输入数据**序列化实操**的输出数据为：

caocao	1447 488	1935
    caopi	400	600	1000
    caozhi	200	300	500
    liubei	215	375	590
    liuxian	850	850	1700
    lvbu 400	600	1000
    sunce	400	600	1000
    sunquan	200	150	350

分析：

定义的Bean对象实现WritableComparable接口，重写compareTo方法

Mapper类中，写出的KV为<bean, 用户名>

Reduce类中，循环输出所有的对象，避免出现总次数相同的情况

public class UserFlowBean implements WritableComparable<UserFlowBean> { 
    
        // 上线次数
        private long upNum;
        // 下线次数
        private long downNum;
        // 总次数
        private long sumNum;
    
        // 空参构造，为了后续反射使用
        public UserFlowBean () { 
    
        }
    
        // 序列化方法
        @Override
        public void write (DataOutput dataOutput) throws IOException { 
    
            dataOutput.writeLong(upNum);
            dataOutput.writeLong(downNum);
            dataOutput.writeLong(sumNum);
        }
    
        // 反序列化方法
        @Override
        public void readFields (DataInput dataInput) throws IOException { 
    
            // 必须要求和序列化方法顺序一致
            upNum = dataInput.readLong();
            downNum = dataInput.readLong();
            sumNum = dataInput.readLong();
        }
    
        @Override
        public String toString () { 
            return upNum + "\t" + downNum + "\t" + sumNum;
        }
    
        // 比较
        @Override
        public int compareTo (UserFlowBean bean) { 
    
            // 比较条件，降序排序
            return Long.compare(bean.getSumNum(), sumNum);
        }
    	...
    }

运行结果如下：

![在这里插入图片描述][20210707113803370.png]

实操（分区排序）：

目的：根据用户姓名进行分区并按照总次数降序排序；

在全排序的基础上加上自定义分区即可。

public class NamePartitioner extends Partitioner<UserFlowBean, Text> { 
    
        @Override
        public int getPartition (UserFlowBean userFlowBean, Text text, int i) { 
            // partition必须从0开始
            int partition = 3;
    
            String key = text.toString();
            if (key.startsWith("cao")) { 
                partition = 0;
            } else if (key.startsWith("liu")) { 
                partition = 1;
            } else if (key.startsWith("sun")) { 
                partition = 2;
            }
    
            return partition;
        }
    }

结果如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]

### Combinber合并 ###

Hadoop常用优化手段之一，能够减少数据的IO传输。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]

自定义Combiner实现的步骤

1、自定义一个 Combiner 继承 Reducer,重写 Reduce 方法；

2、在 Job 驱动类中设置:

job.setCombinerClass(XXXCombiner.class);

实操：

目的：统计过程中对每个MapTask的输出进行局部汇总，以减少网络传输量，这里就可以使用Combiner功能了。

输入数据：

wo shi guanyu
    wo wo wo shi
    guanyu
    yu
    yu
    wo shi liubei de didi
    wo shi guan yunchang
    wo wo wo
    shi shi shi
    guanyu guanyu guanyu

期望Combiner输入数据时多，输出数据时少且经过了合并。

使用**WordCount实操**案例进行改进，改进前的输出过程如下：

![在这里插入图片描述][20210707113836617.png]

Combiner部分代码如下（与原有Reducer完全相同）

public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> { 
    
        IntWritable v = new IntWritable();
    
        @Override
        protected void reduce (Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
    
            // 累加求和
            int sum = 0;
            for (IntWritable value : values) { 
                sum += value.get();
            }
    
            v.set(sum);
    
            context.write(key, v);
        }
    }

在Driver设置使用的Combiner

// 设置Combiner，因为Combiner的代码与Reducer完全相同，因此这里完全可以直接使用原有Reducer来作为Combiner
    job.setCombinerClass(WordCountCombiner.class);

改进后的输出过程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 8]

### 分组排序 ###

GroupingComparator分组排序，也称为辅助排序，对 Reduce 阶段的数据根据某一个或几个字段进行分组。

案例实操：

目的：找出每个订单中最贵的商品

输入数据

订单号   商品号  金额
    0000001 Pdt_01 222.8
    0000002 Pdt_03 522.8
    0000001 Pdt_02 33.8
    0000002 Pdt_04 122.4
    0000002 Pdt_05 722.4
    0000003 Pdt_06 232.8
    0000003 Pdt_02 33.8

期望输出结果

1 222.8
    2 722.4
    4 232.8

分析：

1、利用“订单 id 和成交金额”作为 key,可以将 Map 阶段读取到的所有订单数据按  
照 id 升序排序,如果 id 相同再按照金额降序排序,发送到 Reduce。

2、在 Reduce 端利用 groupingComparator 将订单 id 相同的 kv 聚合成组,然后取第一个即是该订单中最贵商品

MapTask：

Map需要处理的事情：获取数据、分割数据并生成Bean对象；

然后进行二次排序，先根据id升序排序，然后id相同的降序排序。

Bean对象

public class OrderBean implements WritableComparable<OrderBean> { 
    
        private int orderId;
        private double price;
    
        public OrderBean () { 
    
            super();
        }
    
        @Override
        public int compareTo (OrderBean orderBean) { 
    
            // 先按照订单id升序排序，如果相同，按照价格降序排序
            if (orderId > orderBean.getOrderId()) { 
                return 1;
            } else if (orderId < orderBean.getOrderId()) { 
                return -1;
            } else { 
                if (price > orderBean.getPrice()) { 
                    return -1;
                } else if (price < orderBean.getPrice()) { 
                    return 1;
                } else { 
                    return 0;
                }
            }
        }
    
        @Override
        public void write (DataOutput dataOutput) throws IOException { 
    
            dataOutput.writeInt(orderId);
            dataOutput.writeDouble(price);
        }
    
        @Override
        public void readFields (DataInput dataInput) throws IOException { 
    
            orderId = dataInput.readInt();
            price = dataInput.readDouble();
        }
    
        @Override
        public String toString () { 
            return orderId + "\t" + price;
        }
    }

Mapper

// 不需要输出的时候就用NullWritable
    public class GroupingComparatorMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> { 
    
        OrderBean orderBean = new OrderBean();
    
        @Override
        protected void map (LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
    
            // 0000001 Pdt_01 222.8
            System.out.println(key.toString() + ", " + value.toString());
    
            // 1、获取一行数据
            String line = value.toString();
    
            // 2、切割单词
            String[] words = line.split(" ");
    
            // 封装对象
            orderBean.setOrderId(Integer.parseInt(words[0]));
            orderBean.setPrice(Double.parseDouble(words[2]));
    
            context.write(orderBean, NullWritable.get());
        }
    }

Reducer

public class GroupingComparatorReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> { 
    
        NullWritable v = NullWritable.get();
    
        @Override
        protected void reduce (OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { 
    
            context.write(key, NullWritable.get());
            
            // 如果是循环将会输出所有数据，那么输出topX的话就可以进行操作了
            // for (NullWritable value : values) { 
            // context.write(key, NullWritable.get());
            // }
        }
    }

分组排序继承WritableComparator

public class OrderGroupingComparator extends WritableComparator { 
    
        // 必须有无参构造函数
        public OrderGroupingComparator () { 
            // 这里必须要传true，否则会报空指针异常
            super(OrderBean.class, true);
        }
    
        @Override
        public int compare (WritableComparable a, WritableComparable b) { 
    
            // 只要id相同就认为是相同的key，那么就将其分到一组里面
            OrderBean aBean = (OrderBean) a;
            OrderBean bBean = (OrderBean) b;
    
            if (aBean.getOrderId() > bBean.getOrderId()) { 
                return 1;
            } else if (aBean.getOrderId() < bBean.getOrderId()) { 
                return -1;
            } else { 
                return 0;
            }
        }
    }

Driver新增分组排序绑定类

// 设置使用分组排序
    job.setGroupingComparatorClass(OrderGroupingComparator.class);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 9]

分组排序发生在执行reduce方法之前，也就是说先进行分组排序，再进行reduce的输出；宏观上每次是取一组数据之后，将该组数据的进入reduce，然后再取一组进入reduce，直到没有数据；具体流程是进入分组排序时，如果比较的key值第一次相同的时候将会进入reduce。

#### Shuffle总结 ####

##### Map端 #####

当Map 开始产生输出时，它并不是简单的把数据写到磁盘，因为频繁的磁盘操作会导致性能严重下降。它的处理过程更复杂，数据首先是写到内存中的一个缓冲区，并做了一些预排序，以提升效率。

每个Map 任务都有一个用来写入输出数据的循环内存缓冲区。这个缓冲区默认大小是100MB，可以通过**io.sort.mb** 属性来设置具体大小。当缓冲区中的数据量达到一个特定阀值(io.sort.mb \* io.sort.spill.percent，其中**io.sort.spill.percent** 默认是0.80)时，系统将会启动一个后台线程把缓冲区中的内容spill 到磁盘。在spill 过程中，Map 的输出将会继续写入到缓冲区，但如果缓冲区已满，Map 就会被阻塞直到spill 完成。spill 线程在把缓冲区的数据写到磁盘前，会对它进行一个二次快速排序，首先根据数据所属的partition 排序，然后每个partition 中再按Key 排序。输出包括一个索引文件和数据文件。

如果设定了Combiner，将在排序输出的基础上运行。Combiner 就是一个Mini Reducer，它在执行Map 任务的节点本身运行，先对Map 的输出做一次简单Reduce，使得Map 的输出更紧凑，更少的数据会被写入磁盘和传送到Reducer。

spill 文件保存在由**mapred.local.dir**指定的目录中，Map 任务结束后删除。

每当内存中的数据达到spill 阀值的时候，都会产生一个新的spill 文件，所以在Map任务写完它的最后一个输出记录时，可能会有多个spill 文件。在Map 任务完成前，所有的spill 文件将会被归并排序为一个索引文件和数据文件。这是一个多路归并过程，最大归并路数由**io.sort.factor** 控制(默认是10)。如果设定了Combiner，并且spill文件的数量至少是3（由min.num.spills.for.combine 属性控制），那么Combiner 将在输出文件被写入磁盘前运行以压缩数据。

对写入到磁盘的数据进行压缩，通常是一个很好的方法，因为这样做使得数据写入磁盘的速度更快，节省磁盘空间，并减少需要传送到Reducer 的数据量。默认输出是不被压缩的， 但可以很简单的设置**mapred.compress.map.output** 为true 启用该功能。压缩所使用的库由**mapred.map.output.compression.codec** 来设定。

当spill 文件归并完毕后，Map 将删除所有的临时spill 文件，并告知TaskTracker 任务已完成。Reducers 通过HTTP来获取对应的数据。用来传输partitions 数据的工作线程数由**tasktracker.http.threads** 控制，这个设定是针对每一个TaskTracker 的，并不是单个Map，默认值为40，在运行大作业的大集群上可以增大以提升数据传输速率。

##### Reduce端 #####

###### copy阶段 ######

Map 的输出文件放置在运行Map 任务的TaskTracker 的本地磁盘上（注意：Map 输出总是写到本地磁盘，但Reduce 输出不是，一般是写到HDFS），它是运行Reduce 任务的TaskTracker 所需要的输入数据。Reduce 任务的输入数据分布在集群内的多个Map 任务的输出中，Map 任务可能会在不同的时间内完成，只要完成的Map 任务数达到占总Map任务数一定比例（**mapred.reduce.slowstart.completed.maps** 默认0.05），Reduce 任务就开始拷贝它的输出。

Reduce 任务拥有多个拷贝线程， 可以并行的获取Map 输出。可以通过设定**mapred.reduce.parallel.copies** 来改变线程数，默认是5。

如果Map 输出足够小，它们会被拷贝到Reduce TaskTracker 的内存中（缓冲区的大小

由**mapred.job.shuffle.input.buffer.percent** 控制，指定了用于此目的的堆内存的百分比）；如果缓冲区空间不足，会被拷贝到磁盘上。当内存中的缓冲区用量达到一定比例阀值（由**mapred.job.shuffle.merge.percent** 控制），或者达到了Map 输出的阀值大小（由**mapred.inmem.merge.threshold** 控制），缓冲区中的数据将会被归并然后spill到磁盘。

拷贝来的数据叠加在磁盘上，有一个后台线程会将它们归并为更大的排序文件，这样做节省了后期归并的时间。对于经过压缩的Map 输出，系统会自动把它们解压到内存方便对其执行归并。

###### sort阶段 ######

当所有的Map 输出都被拷贝后，Reduce 任务进入排序阶段（更恰当的说应该是归并阶段，因为排序在Map 端就已经完成），这个阶段会对所有的Map 输出进行归并排序，这个工作会重复多次才能完成。

假设这里有50 个Map 输出（可能有保存在内存中的），并且归并因子是10（由**io.sort.factor** 控制，就像Map 端的merge 一样），那最终需要5 次归并。每次归并会把10个文件归并为一个，最终生成5 个中间文件。

注：每趟合并的文件数实际上比示例中展示的更微妙。目标是合并最小数量的文件以便满足最后一趟的合并系数。因此如果是40个文件，我们不会在四趟中，每趟合并10个文件从而得到4个文件。相反，第一趟只合并4个文件，随后三趟合并所有十个文件。在最后一趟中，4个已合并的文件和余下的6个（未合并的）文件合计10个文件。这并没有改变合并的次数，它只是一个优化措施，尽量减少写到磁盘的数据量，因为最后一趟总是直接合并到reduce。

###### reduce阶段 ######

在Reduce 阶段，Reduce 函数会作用在排序输出的每一个key 上。这个阶段的输出被直接写到输出文件系统，一般是HDFS。在HDFS 中，因为TaskTracker 节点也运行着一个DataNode 进程，所以第一个块备份会直接写到本地磁盘。

#### Shuffle配置调优 ####

Shuffle是MapReduce中最耗时的阶段，因为涉及大量的IO操作。

该配置调优方案主要是对以上Shuffle整个过程中涉及到的配置项按流程顺序一一呈现并给以调优建议。

##### Map端 #####

1.  io.sort.mb

用于map输出排序的内存缓冲区大小

类型：Int

默认：100mb

备注：如果能估算map输出大小，就可以合理设置该值来尽可能减少溢出写的次数，这对调优很有帮助。

2)io.sort.spill.percent

map输出排序时的spill阀值（即使用比例达到该值时，将缓冲区中的内容spill 到磁盘）

类型：float

默认：0.80

3)io.sort.factor

归并因子（归并时的最多合并的流数），map、reduce阶段都要用到

类型：Int

默认：10

备注：将此值增加到100是很常见的。

4)min.num.spills.for.combine

运行combiner所需的最少溢出写文件数（如果已指定combiner）

类型：Int

默认：3

5)mapred.compress.map.output

map输出是否压缩

类型：Boolean

默认：false

备注：如果map输出的数据量非常大，那么在写入磁盘时压缩数据往往是个很好的主意，因为这样会让写磁盘的速度更快，节约磁盘空间，并且减少传给reducer的数据量。

6)mapred.map.output.compression.codec

用于map输出的压缩编解码器

类型：Classname

默认：org.apache.[Hadoop][].io.compress.DefaultCodec

备注：推荐使用LZO压缩。Intel内部测试表明，相比未压缩，使用LZO压缩的 TeraSort作业，运行时间减少60%，且明显快于Zlib压缩。

1.  tasktracker.http.threads

每个tasktracker的工作线程数，用于将map输出到reducer。

（注：这是集群范围的设置，不能由单个作业设置）

类型：Int

默认：40

备注：tasktracker开http服务的线程数。用于reduce拉取map输出数据，大集群可以将其设为40~50。

##### reduce端 #####

1)mapred.reduce.slowstart.completed.maps

调用reduce之前，map必须完成的最少比例

类型：float

默认：0.05

2)mapred.reduce.parallel.copies

reducer在copy阶段同时从mapper上拉取的文件数

类型：int

默认：5

3)mapred.job.shuffle.input.buffer.percent

在shuffle的复制阶段，分配给map输出的缓冲区占堆空间的百分比

类型：float

默认：0.70

4)mapred.job.shuffle.merge.percent

map输出缓冲区（由mapred.job.shuffle.input.buffer.percent定义）使用比例阀值，当达到此阀值，缓冲区中的数据将会被归并然后spill 到磁盘。

类型：float

默认：0.66

5)mapred.inmem.merge.threshold

map输出缓冲区中文件数

类型：int

默认：1000

备注：0或小于0的数意味着没有阀值限制，溢出写将有mapred.job.shuffle.merge.percent单独控制。

6)mapred.job.reduce.input.buffer.percent

在reduce过程中，在内存中保存map输出的空间占整个堆空间的比例。

类型：float

默认：0.0

备注：reduce阶段开始时，内存中的map输出大小不能大于该值。默认情况下，在reduce任务开始之前，所有的map输出都合并到磁盘上，以便为reducer提供尽可能多的内存。然而，如果reducer需要的内存较少，则可以增加此值来最小化访问磁盘的次数，以提高reduce性能。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20221005/73724c158253462c9bd55cf1a17dd2b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20221005/4e3c47240f60416a9ee2b7514508ca78.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]: /images/20221005/2159a0ddcacb4e1c87aba9e613d9b0db.png
[20210707113652272.png]: /images/20221005/49aba09721164d74b928071c97af0ef0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]: /images/20221005/0982f40aeb6d434a940cc95b655b03ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]: /images/20221005/f534d01ff9a44f1283e08f8159a0f28b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]: /images/20221005/f586f59626ab469cba6bc11dca22ea07.png
[20210707113803370.png]: /images/20221005/cb64e55ff59740a6bc03b710e784ad94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]: /images/20221005/bbe8cbc5d73e48a8b70aa1cc3fe2656a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]: /images/20221005/b7c483c54c11484891df24c783d6f10a.png
[20210707113836617.png]: /images/20221005/8f4c90fd7e7d4517966fcf1f2791ab85.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 8]: /images/20221005/68fe5afa177a47dda7fa45e12371c3a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 9]: /images/20221005/5bd20be0ed43484f92936837236e735a.png
[Hadoop]: http://lib.csdn.net/base/20