理解MapReduce思想

需求：有一个五层的图书馆，需要获取图书馆中一共有多少本书。

只有一个人时，是能一本一本的数！工作量巨大，耗时较长。

分配五个人由你支配。此时你怎么支配？

五个人，每个人数一层的书量，最终将五个人的量汇总求和，就是图书馆中最终书的数量。

 

MapReduce的思想核心是“分而治之，先分后合”。即将一个大的、复杂的工作或任务，拆分成多个小的任务，并行处理，最终进行合并。适用于大量复杂的、时效性不高的任务处理场景（大规模离线数据处理场景）。即使是发布过论文实现分布式计算的谷歌也只是实现了这种思想，而不是自己原创。

MapReduce由两部分组成，分别是Map 和Reduce两部分。

Map负责“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系。例如前面例子中的分配每个人数一层楼。

Reduce负责“合”，即对map阶段的结果进行全局汇总。例如前面例子中将五个人的结果汇总。

这两个阶段合起来正是MapReduce思想的体现。

 

MapReduce并行计算

HDFS存储数据时对大于128M的数据会进行数据切分，每128M一个数据块，数据块会分散、分布存储到HDFS。

MapReduce在进行计算前会复制计算程序，每个数据块会分配一个独立的计算程序副本（MapTack）。计算时多个数据块几乎同时被读取并计算，但是计算程序完全相同。最终将各个计算程序计算的结果进行汇总。

 

Hadoop -MapReduce设计构思

MapReduce是一个分布式运算程序的编程框架，核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在Hadoop集群上。

既然是做计算的框架，那么表现形式就是有个输入（input），MapReduce操作这个输入（input），通过本身定义好的计算模型，得到一个输出（output）。

Hadoop MapReduce构思体现在如下的三个方面：

如何应对大数据处理：分而治之：对相互间不具有计算依赖关系的大数据，实现并行最自然的办法就是采取分而治之的策略。并行计算的第一个重要问题是如何划分计算任务或者计算数据以便对划分的子任务或数据块同时进行计算。不可分拆的计算任务或相互间有依赖关系的数据无法进行并行计算

构建抽象模型：Map和Reduce：   MapReduce借鉴了函数式语言中的思想，用Map和Reduce两个函数提供了高层的并行编程抽象模型。Map: 对一组数据元素进行某种重复式的处理

Map: 对一组数据元素进行某种重复式的处理

Reduce: 对Map的中间结果进行某种进一步的结果整理。

MapReduce中定义了如下的Map和Reduce两个抽象的编程接口，由用户去编程实现：

map: [k1,v1] → [(k2,v2)]

reduce: [k2, {v2,…}] → [k3, v3]

 

WordCount体现每个KeyValue

Map和Reduce为程序员提供了一个清晰的操作接口抽象描述。通过以上两个编程接口，大家可以看出MapReduce处理的数据类型是<key,value>键值对。

 

MapReduce编程初体验

举个例子：在给定的文本文件中统计输出每一个单词出现的总次数（在本地中做输入输出）

准备数据：

hello,world,hadoop

hello,hive,sqoop,flume

kitty,tom,jerry,world

hadoop

 

POM文件：

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

 

    <groupId>cn.itcast</groupId>

    <artifactId>mapreduce</artifactId>

    <version>1.0-SNAPSHOT</version>

    <repositories>

        <repository>

            <id>cloudera</id>

            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>

        </repository>

    </repositories>

    <dependencies>

        <dependency>

            <groupId>org.apache.Hadoop</groupId>

            <artifactId>Hadoop-client</artifactId>

            <version>2.6.0-mr1-cdh5.14.0</version>

        </dependency>

        <dependency>

            <groupId>org.apache.Hadoop</groupId>

            <artifactId>Hadoop-common</artifactId>

            <version>2.6.0-cdh5.14.0</version>

        </dependency>

        <dependency>

            <groupId>org.apache.Hadoop</groupId>

            <artifactId>Hadoop-hdfs</artifactId>

            <version>2.6.0-cdh5.14.0</version>

        </dependency>

 

        <dependency>

            <groupId>org.apache.Hadoop</groupId>

            <artifactId>Hadoop-mapreduce-client-core</artifactId>

            <version>2.6.0-cdh5.14.0</version>

        </dependency>

        <dependency>

            <groupId>junit</groupId>

            <artifactId>junit</artifactId>

            <version>4.11</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.testng</groupId>

            <artifactId>testng</artifactId>

            <version>RELEASE</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-compiler-plugin</artifactId>

                <version>3.0</version>

                <configuration>

                    <source>1.8</source>

                    <target>1.8</target>

                    <encoding>UTF-8</encoding>

                </configuration>

            </plugin>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-shade-plugin</artifactId>

                <version>2.4.3</version>

                <executions>

                    <execution>

                        <phase>package</phase>

                        <goals>

                            <goal>shade</goal>

                        </goals>

                        <configuration>

                            <minimizeJar>true</minimizeJar>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

 

        </plugins>

    </build>

</project>

 

WordCountMapper 类：

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

 

public class WordCountMapper extends Mapper<LongWritable,Text,Text,LongWritable> {
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] split = line.split(",");
        for (String word : split) {
            context.write(new Text(word),new LongWritable(1));
        }

    }
}

 

WordCountReducer 类：

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

 

public class WordCountReducer extends Reducer<Text,LongWritable,Text,LongWritable> {
    /**
     * 自定义reduce逻辑
     * 所有的key都是单词，所有的values都是单词出现的次数
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        long count = 0;
        for (LongWritable value : values) {
            count += value.get();
        }
        context.write(key,new LongWritable(count));
    }
}

 

主类JobMain：

public class JobMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(),        
        JobMain.class.getSimpleName());
        job.setJarByClass(JobMain.class); 
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new       
        Path("hdfs://192.168.100.129:8020/wordcount"));
        job.setMapperClass(WordCountMapper.class); 
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class); 
        job.setReducerClass(WordCountReducer.class); 
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new
        Path("hdfs://192.168.100.129:8020/wordcount_out"));
        boolean b = job.waitForCompletion(true);
        return b?0:1;
    }

    /**
     * 程序main函数的入口类
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        Tool tool  =  new JobMain();
        int run = ToolRunner.run(configuration, tool, args);
        System.exit(run);
    }
}

 

如果说运行时报这个错误：

Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.AccessControlException): Permission denied: user=admin, access=WRITE, inode="/":root:supergroup:drwxr-xr-x

直接在hdfs-site.xml中把权限控制关闭就行了

<property>

                <name>dfs.permissions</name>

                <value>false</value>

   </property>

 

然后重启hdfs的集群

代码编写完毕后，将代码打成jar包放到服务器上去运行：

hadoop jar hadoop_hdfs_operate-1.0-SNAPSHOT.jar  JobMain主类的相对路径

 

运行过程：

运行结果：

 

 

如果你觉得这篇内容对你挺有启发：

点赞，让更多的人也能看到这篇内容（看完不点赞，都是耍流氓 -_-）

不定期分享原创知识。。。。嘿嘿