【分布式计算框架】 MapReduce编程初级实践

简介: 【分布式计算框架】 MapReduce编程初级实践

MapReduce编程初级实践

一、实验目的

  • 编程WordCount
  • 编程实现文件合并和去重操作
  • 编程实现对输入文件的排序

二、实验环境

  • centos 6.5
  • VMware Workstation

三、实验内容

mapreduce高可用环境配置

伪分布式(单节点)修改配置:

(1) mapred-site.xml

 <property>
     <name>mapreduce.framework.name</name>
     <value>yarn</value>
 </property>

(2) yarn-site.xml

 <property>
     <name>yarn.nodemanager.aux-services</name>
     <value>mapreduce_shuffle</value>
 </property>

复制目录D:\software\eclipce\workspace\Test20191909\src

cd /opt/20191909/hadoop-2.6.5/etc/hadoop 进入配置文件的目录

ll 查看文件

将yarn-site.xml和mapred-site.xml文件复制到D:\software\eclipce\workspace\Test20191909\src

新建一个类

启动resourcemanager服务

编程WordCount

(1)创建一个新文件

for i in `seq 100000`;do echo "hello jxxy$i" >> test.txt;done

(2)编程MyWC主类,MyMapper类,MyReducer类,制作jar包

  • MyWC主类
package com.sxt.mr.wc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MyWC {
  public static void main(String[] args){
     Configuration conf=new Configuration();        

         try{

         //创建一个新作业
         Job job=Job.getInstance(conf);
         job.setJarByClass(MyWC.class); //jar包      
         job.setJobName("myjob");
       
         //job.setInputPath(new Path());
         //job.setOutputPath(new Path());
         Path inPath=new Path("/user/root/test.txt");
         FileInputFormat.addInputPath(job,inPath);
         //org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
        
         Path outPath=new Path("/output/wordcount");
         //如果输出路径存在,则先删除
         if(outPath.getFileSystem(conf).exists(outPath))
            outPath.getFileSystem(conf).delete(outPath,true);
         FileOutputFormat.setOutputPath(job,outPath); 
         //org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

         //创建MyMapper,MyReducer两个类 
         job.setMapperClass(MyMapper.class);
         job.setMapOutputKeyClass(Text.class);
         job.setOutputValueClass(IntWritable.class);
         job.setReducerClass(MyReducer.class);
       
          //提交作业
         job.waitForCompletion(true);

          }
          catch(Exception e){
          }

  }
}

  • MyMapper类
package com.sxt.mr.wc;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper extends Mapper<Object,Text,Text,IntWritable> {
    private final static IntWritable one=new IntWritable(1);
    private Text word=new Text();

    public void map(Object key,Text value,Context context) throws IOException,InterruptedException{
       StringTokenizer str=new StringTokenizer(value.toString());
       while(str.hasMoreTokens()){
          word.set(str.nextToken());
          context.write(word,one);
       }
    }
    }


  • MyReducer类
package com.sxt.mr.wc;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MyReducer extends Reducer<Text,IntWritable,Text,IntWritable>{

//迭代计算
private IntWritable result=new IntWritable();

public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{
   int sum=0;
   for(IntWritable val:values){
      sum+=val.get();
   }
   result.set(sum);
   context.write(key,result);
}
}

  • 打成jar包

将jar包上传

在集群上执行命令

hadoop jar wc.jar com.sxt.mr.wc.MyWC

(3)运行程序,统计test.txt文件hello和jxxy出现的次数

查看浏览器8088端口

编程实现文件合并和去重操作

对于两个输入文件,即文件A和文件B,编写程序对两个文件进行合并,并剔除其中重复的内容,

得到一个新的输出文件C。

样例如下:

文件A

20150101 x
20150102 y
20150103 x
20150104 y
20150105 z
20150106 x

文件B

20150101 y
20150102 y
20150103 x
20150104 z
20150105 y

合并如下

20150101 x  
20150101 y  
20150102 y  
20150103 x  
20150104 y  
20150104 z  
20150105 y  
20150105 z  
20150106 x  

代码如下

FileMergeDriver类

package com.sxt.file.test1;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FileMergeDriver extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), "FileMerge");
        job.setJarByClass(FileMergeDriver.class);

        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileInputFormat.addInputPath(job, new Path(args[1]));
        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new FileMergeDriver(), args);
        System.exit(exitCode);
    }
}


MyReducer类

package com.sxt.file.test1;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MyReducer extends Reducer<Text, Text, Text, Text> {
    private final static Text nonDuplicate = new Text();

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        nonDuplicate.set(""); // 设置空值
        context.write(key, nonDuplicate); // 将key写入输出,实现去重
    }
}

MyMapper类

package com.sxt.file.test1;

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

import java.io.IOException;

public class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
    private final static Text nonDuplicate = new Text();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        nonDuplicate.set(""); // 设置空值
        context.write(value, nonDuplicate); // 将每行文本作为key输出
    }
}

将jar包上传

准备测试数据

A.txt

B.txt

执行命令

hdfs dfs -put ./A.txt /user/root
hdfs dfs -put ./B.txt /user/root
hadoop jar test1.jar com.sxt.file.test1.FileMergeDriver /user/root/A.txt /user/root/B.txt /user/root/C.txt

查看结果与样例一致实验成功

编程实现对输入文件的排序

现在有多个输入文件,每个文件中的每行内容均为一个整数。要求读取所有文件中的整数,进行升序排序后,输出到一个新的

文件中,输出的整数格式为每行两个整数,第一个整数位第二个整数的排序位次,第二个整数位原待排列的整数。

样例如下:

文件1

33
37
12
40

文件2

4
16
39
5

文件3

1
45
25

输出文件

1 1
2 4
3 5
4 12
5 16
6 25
7 33
8 37
9 39
10 40
11 45

代码如下

SortIntegers类

package com.sxt.file.test2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SortIntegers {
    public static void main(String[] args) {
        try {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf, "sort integers job");
            job.setJarByClass(SortIntegers.class);
            job.setMapperClass(SortIntegersMapper.class);
            job.setReducerClass(SortIntegersReducer.class);
            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.addInputPath(job, new Path(args[0])); // 输入文件路径
            FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出文件路径
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


SortIntegersReducer类

package com.sxt.file.test2;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class SortIntegersReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
    private IntWritable rank = new IntWritable(1);
    private IntWritable number = new IntWritable();

    public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        for (IntWritable val : values) {
            number.set(key.get());
            context.write(rank, number);
            rank.set(rank.get() + 1);
        }
    }
}

SortIntegersMapper类

package com.sxt.file.test2;

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

public class SortIntegersMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
    private final static IntWritable lineNumber = new IntWritable(1);
    private IntWritable number = new IntWritable();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        int num = Integer.parseInt(line.trim());
        number.set(num);
        context.write(number, lineNumber);
        lineNumber.set(lineNumber.get() + 1);
    }
}

打成jar包

上传jar包

创建测试文件

1.txt

2.txt

3.txt

将文件上传到HDFS

hadoop fs -mkdir /user/root/sort_test
hdfs dfs -put ./1.txt /user/root/sort_test
hdfs dfs -put ./2.txt /user/root/sort_test
hdfs dfs -put ./3.txt /user/root/sort_test
hadoop jar test2.jar com.sxt.file.test2.SortIntegers /user/root/sort_test /user/root/sort_test/result

实验结果与样例一致,实验成功

四、出现的问题及解决方案

  1. DFS Location拒绝连接

    解决方案:

杀死所有java进程后开启集群

start-dfs.sh

start-yarn.sh

无效

新建一个Location,把端口号改为9000(这里是伪分布式的端口,之前没有修改,也是错误的原因)

解决


执行上传作业,版本不一致

解决方案

在 Eclipse 中选择要修改的项目,然后右键单击该项目,在弹出菜单中选择 “Properties”(属性)。

  1. 在弹出的对话框中,找到并展开 “Java Build Path”(Java 构建路径)选项。
  2. 在 “Java Build Path” 下,点击 “Libraries”(库),然后点击 “JRE System Library”。
  3. 点击 “Edit”(编辑)按钮,然后选择合适的 JRE(Java 运行时环境)版本。可以选择系统中已安装的 JRE 版本,也可以选择其他已配置的 JRE。
  4. 点击 “Finish”(完成)保存更改。
  5. 接下来,还需要修改项目的编译级别:
  • 在项目属性对话框中,找到 “Java Compiler”(Java 编译器)选项。
  • 确保选中 “Enable project specific settings”(启用项目特定设置)复选框。
  • 在 “Compiler compliance level”(编译器兼容性级别)下拉菜单中选择你想要的 Java 版本。
  • 点击 “Apply and Close”(应用并关闭)保存更改。

五、实验结果

运行程序,统计test.txt文件hello和jxxy出现的次数

运行程序,实现文件合并和去重操作

运行程序,实现对输入文件的排序

六、实验思考题

  1. 完全分布式配置哪些文件?
  • mapred-site.xml
  • yarn-site.xml
  1. 试述MapReduce的工作流程。

MapReduce是一种用于处理大规模数据集的并行计算编程模型。其工作流程包括以下几个步骤:

  1. 划分阶段(Input Split)
  • 输入数据集被划分成若干个输入切片(input splits),每个输入切片会被一个Map任务处理。
  1. 映射阶段(Map Stage)
  • 每个Map任务读取一个输入切片,并对其进行处理,生成中间键值对(key-value pairs)。
  • 中间键值对由用户自定义的Map函数生成,这些键值对会被分区函数划分到不同的Reducer任务。
  1. 分区和排序阶段(Shuffle and Sort Stage)
  • 中间键值对根据键进行排序,并按照分区函数的规则划分到不同的Reducer任务。
  • 分区函数的作用是确保相同键的键值对会被发送到同一个Reducer任务。
  1. 归约阶段(Combine Stage,可选)
  • 可选的归约(Combiner)函数可以在Map阶段后执行,用于局部聚合中间键值对,以减少数据传输量。
  1. 合并阶段(Reduce Stage)
  • 每个Reduce任务接收来自Map任务输出的中间键值对,并按照键进行分组。
  • Reduce任务对每个键的值列表执行用户定义的Reduce函数,生成最终的输出结果。
  1. 输出阶段(Output Stage)
  • 最终的输出结果会被写入HDFS或其他存储系统,用于进一步的分析或处理。
  • 输入数据集被划分成若干个输入切片(input splits),每个输入切片会被一个Map任务处理。

整个MapReduce过程涉及到数据的划分、映射、分区、排序、归约和合并等操作,通过并行化处理大规模数据集,实现高效的数据处理和分析。 MapReduce的工作流程简化了分布式计算任务的编写和执行,使得处理海量数据变得更加容易和高效。

相关文章
|
7天前
|
负载均衡 监控 Go
使用Golang框架构建分布式系统
本文探讨了使用Golang构建分布式系统的方法。Golang因其高效、简洁的语法和并发支持成为理想的开发语言。文中列举了几个常用的Golang框架,如Echo、Gin、gRPC和NATS等,并强调了服务拆分、通信机制、负载均衡等构建分布式系统的关键要素。通过选择合适的框架,遵循需求分析、技术选型、服务设计等步骤,开发者可以构建出高性能、高可用和可扩展的系统。此外,文中还提供了一个使用gRPC和etcd的简单代码案例来说明实现过程。
|
4天前
|
存储 分布式计算 Hadoop
Hadoop生态系统详解:HDFS与MapReduce编程
Apache Hadoop是大数据处理的关键,其核心包括HDFS(分布式文件系统)和MapReduce(并行计算框架)。HDFS为大数据存储提供高容错性和高吞吐量,采用主从结构,通过数据复制保证可靠性。MapReduce将任务分解为Map和Reduce阶段,适合大规模数据集的处理。通过代码示例展示了如何使用MapReduce实现Word Count功能。HDFS和MapReduce的结合,加上YARN的资源管理,构成处理和分析大数据的强大力量。了解和掌握这些基础对于有效管理大数据至关重要。【6月更文挑战第12天】
|
5天前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
11 1
|
6天前
|
分布式计算 自然语言处理 大数据
【大数据】MapReduce JAVA API编程实践及适用场景介绍
【大数据】MapReduce JAVA API编程实践及适用场景介绍
16 0
|
7天前
|
分布式计算 负载均衡 并行计算
Python 分布式计算框架 PP (Parallel Python):集群模式下的实践探索
该文介绍了使用Parallel Python (PP) 在两台物理机上构建分布式计算集群的经验。PP是一个轻量级框架,旨在简化Python代码在多处理器系统和集群中的并行执行。文中通过设置子节点的IP、端口和密钥启动PP服务器,并在主节点创建PP实例进行负载均衡。实验使用官方的质数和计算示例,显示PP在集群模式下能有效利用多台机器的多核CPU,实现计算效率的显著提升。未来,作者计划进一步研究PP在更复杂任务和大规模集群中的应用潜力。
|
1月前
|
NoSQL Java 关系型数据库
【Redis系列笔记】分布式锁
分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。 分布式锁的核心思想就是让大家都使用同一把锁,只要大家使用的是同一把锁,那么我们就能锁住线程,不让线程进行,让程序串行执行,这就是分布式锁的核心思路
405 2
|
4天前
|
NoSQL 算法 Java
探讨redis分布式锁
探讨redis分布式锁
10 1
|
11天前
|
缓存 NoSQL 安全
玩转Redis!非常强大的Redisson分布式集合,少写60%代码
Redisson是Java的Redis客户端,提供实时数据平台服务,简化了分布式环境下的数据管理。它包含RList、RSet、RMap等分布式集合,支持ConcurrentMap和Set接口,确保线程安全和数据一致性。例如,RMap实现了本地缓存和监听器功能,允许数据监听和本地加速读取。此外,还提供了RSet的排序和去重功能,以及RQueue和RBlockingQueue等队列实现,支持阻塞操作。通过Redisson,开发者能轻松处理分布式系统的数据同步和操作。
|
13天前
|
存储 缓存 NoSQL
了解Redis,第一弹,什么是RedisRedis主要适用于分布式系统,用来用缓存,存储数据,在内存中存储那么为什么说是分布式呢?什么叫分布式什么是单机架构微服务架构微服务的本质
了解Redis,第一弹,什么是RedisRedis主要适用于分布式系统,用来用缓存,存储数据,在内存中存储那么为什么说是分布式呢?什么叫分布式什么是单机架构微服务架构微服务的本质
|
15天前
|
缓存 NoSQL 关系型数据库
【Redis】 浅谈分布式架构
【Redis】 浅谈分布式架构

热门文章

最新文章