热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

开发笔记:HadoopMapreduce

代码如下:

代码如下:

第一个mapper:FindFriendMapTaskByOne

 


package com.gec.demo;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
import java.io.PrintStream;
public class FindFriendMapTaskByOne extends Mapper {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line=value.toString();
String[] datas=line.split(":");
String user=datas[0];
String []friends=datas[1].split(",");
for (String friend : friends) {
context.write(new Text(friend),new Text(user));
}
}
}

第一个reducer:


package com.gec.demo;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class FindFriendReducerTaskByOne extends Reducer {
@Override
protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
StringBuffer strBuf=new StringBuffer();
for (Text value : values) {
strBuf.append(value).append("-");
}
context.write(key,new Text(strBuf.toString()));
}
}

第一个job


package com.gec.demo;
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;
import java.io.IOException;
public class FindFriendJobByOne {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration cOnfiguration=new Configuration();
Job job=Job.getInstance(configuration);
//设置Driver类
job.setJarByClass(FindFriendJobByOne.class);
//设置运行那个map task
job.setMapperClass(FindFriendMapTaskByOne .class);
//设置运行那个reducer task
job.setReducerClass(FindFriendReducerTaskByOne .class);
//设置map task的输出key的数据类型
job.setMapOutputKeyClass(Text.class);
//设置map task的输出value的数据类型
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//指定要处理的数据所在的位置
FileInputFormat.setInputPaths(job, "D://Bigdata//4、mapreduce//day05//homework//friendhomework.txt");
//指定处理完成之后的结果所保存的位置
FileOutputFormat.setOutputPath(job, new Path("D://Bigdata//4、mapreduce//day05//homework//output"));
//向yarn集群提交这个job
boolean res = job.waitForCompletion(true);
System.exit(res?0:1);
}
}

得出结果:

第二个mapper:


package com.gec.demo;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class FindFriendMapTaskByTwo extends Mapper {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line=value.toString();
String []datas=line.split("\\t");
String []userlist=datas[1].split("-");
for (int i=0;i){
for (int j=i+1;j){
String user1=userlist[i];
String user2=userlist[j];
String friendkey=user1+"-"+user2;
context.write(new Text(friendkey),new Text(datas[0]));
}
}
}
}

第二个reducer:


package com.gec.demo;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class FindFriendReducerTaskByTwo extends Reducer {
@Override
protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
StringBuffer stringBuffer=new StringBuffer();
for (Text value : values) {
stringBuffer.append(value).append(",");
}
context.write(key,new Text(stringBuffer.toString()));
}
}

第二个job:


package com.gec.demo;
import org.apache.hadoop.conf.Configuration;
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 java.io.IOException;
public class FindFriendJobByTwo {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration cOnfiguration=new Configuration();
Job job=Job.getInstance(configuration);
//设置Driver类
job.setJarByClass(FindFriendJobByTwo.class);
//设置运行那个map task
job.setMapperClass(FindFriendMapTaskByTwo .class);
//设置运行那个reducer task
job.setReducerClass(FindFriendReducerTaskByTwo .class);
//设置map task的输出key的数据类型
job.setMapOutputKeyClass(Text.class);
//设置map task的输出value的数据类型
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//指定要处理的数据所在的位置
FileInputFormat.setInputPaths(job, "D://Bigdata//4、mapreduce//day05//homework//friendhomework3.txt");
//指定处理完成之后的结果所保存的位置
FileOutputFormat.setOutputPath(job, new Path("D://Bigdata//4、mapreduce//day05//homework//output"));
//向yarn集群提交这个job
boolean res = job.waitForCompletion(true);
System.exit(res?0:1);
}
}

得出结果:


案例四


MapReduce中多表合并案例

1)需求:

订单数据表t_order:

























id


pid


amount


1001


01


1


1002


02


2


1003


03


3


 

商品信息表t_product





















id


pname


01


小米


02


华为


03


格力


 

       将商品信息表中数据根据商品id合并到订单数据表中。

最终数据形式:








































id


pname


amount


1001


小米


1


1001


小米


1


1002


华为


2


1002


华为


2


1003


格力


3


1003


格力


3



3.4.1 需求1:reduce端表合并(数据倾斜)

通过将关联条件作为map输出的key,将两表满足join条件的数据并携带数据所来源的文件信息,发往同一个reduce task,在reduce中进行数据的串联。

 

 

1)创建商品和订合并后的bean类








package com.gec.mapreduce.table;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.Writable;

 

public class TableBean implements Writable {

       private String order_id; // 订单id

       private String p_id; // 产品id

       private int amount; // 产品数量

       private String pname; // 产品名称

       private String flag;// 表的标记

 

       public TableBean() {

              super();

       }

 

       public TableBean(String order_id, String p_id, int amount, String pname, String flag) {

              super();

              this.order_id = order_id;

              this.p_id = p_id;

              this.amount = amount;

              this.pname = pname;

              this.flag = flag;

       }

 

       public String getFlag() {

              return flag;

       }

 

       public void setFlag(String flag) {

              this.flag = flag;

       }

 

       public String getOrder_id() {

              return order_id;

       }

 

       public void setOrder_id(String order_id) {

              this.order_id = order_id;

       }

 

       public String getP_id() {

              return p_id;

       }

 

       public void setP_id(String p_id) {

              this.p_id = p_id;

       }

 

       public int getAmount() {

              return amount;

       }

 

       public void setAmount(int amount) {

              this.amount = amount;

       }

 

       public String getPname() {

              return pname;

       }

 

       public void setPname(String pname) {

              this.pname = pname;

       }

 

       @Override

       public void write(DataOutput out) throws IOException {

              out.writeUTF(order_id);

              out.writeUTF(p_id);

              out.writeInt(amount);

              out.writeUTF(pname);

              out.writeUTF(flag);

       }

 

       @Override

       public void readFields(DataInput in) throws IOException {

              this.order_id = in.readUTF();

              this.p_id = in.readUTF();

              this.amount = in.readInt();

              this.pname = in.readUTF();

              this.flag = in.readUTF();

       }

 

       @Override

       public String toString() {

              return order_id + "\\t" + p_id + "\\t" + amount + "\\t" ;

       }

}


2)编写TableMapper程序








package com.gec.mapreduce.table;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.lib.input.FileSplit;

 

public class TableMapper extends Mapper{

       TableBean bean = new TableBean();

       Text k = new Text();

      

       @Override

       protected void map(LongWritable key, Text value, Context context)

                     throws IOException, InterruptedException {

             

              // 1 获取输入文件类型

              FileSplit split = (FileSplit) context.getInputSplit();

              String name = split.getPath().getName();

             

              // 2 获取输入数据

              String line = value.toString();

             

              // 3 不同文件分别处理

              if (name.startsWith("order")) {// 订单表处理

                     // 3.1 切割

                     String[] fields = line.split(",");

                    

                     // 3.2 封装bean对象

                     bean.setOrder_id(fields[0]);

                     bean.setP_id(fields[1]);

                     bean.setAmount(Integer.parseInt(fields[2]));

                     bean.setPname("");

                     bean.setFlag("0");

                    

                     k.set(fields[1]);

              }else {// 产品表处理

                     // 3.3 切割

                     String[] fields = line.split(",");

                    

                     // 3.4 封装bean对象

                     bean.setP_id(fields[0]);

                     bean.setPname(fields[1]);

                     bean.setFlag("1");

                     bean.setAmount(0);

                     bean.setOrder_id("");

                    

                     k.set(fields[0]);

              }

              // 4 写出

              context.write(k, bean);

       }

}


3)编写TableReducer程序








package com.gec.mapreduce.table;

import java.io.IOException;

import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class TableReducer extends Reducer {

 

       @Override

       protected void reduce(Text key, Iterable values, Context context)

                     throws IOException, InterruptedException {

 

              // 1准备存储订单的集合

              ArrayList orderBeans = new ArrayList<>();

              // 2 准备bean对象

              TableBean pdBean = new TableBean();

 

              for (TableBean bean : values) {

 

                     if ("0".equals(bean.getFlag())) {// 订单表

                            // 拷贝传递过来的每条订单数据到集合中

                            TableBean orderBean = new TableBean();

                            try {

                                   BeanUtils.copyProperties(orderBean, bean);

                            } catch (Exception e) {

                                   e.printStackTrace();

                            }

 

                            orderBeans.add(orderBean);

                     } else {// 产品表

                            try {

                                   // 拷贝传递过来的产品表到内存中

                                   BeanUtils.copyProperties(pdBean, bean);

                            } catch (Exception e) {

                                   e.printStackTrace();

                            }

                     }

              }

 

              // 3 表的拼接

              for(TableBean bean:orderBeans){

                     bean.setP_id(pdBean.getPname());

                    

                     // 4 数据写出去

                     context.write(bean, NullWritable.get());

              }

       }

}


4)编写TableDriver程序








package com.gec.mapreduce.table;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.NullWritable;

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 TableDriver {

 

       public static void main(String[] args) throws Exception {

              // 1 获取配置信息,或者job对象实例

              Configuration cOnfiguration= new Configuration();

              Job job = Job.getInstance(configuration);

 

              // 2 指定本程序的jar包所在的本地路径

              job.setJarByClass(TableDriver.class);

 

              // 3 指定本业务job要使用的mapper/Reducer业务类

              job.setMapperClass(TableMapper.class);

              job.setReducerClass(TableReducer.class);

 

              // 4 指定mapper输出数据的kv类型

              job.setMapOutputKeyClass(Text.class);

              job.setMapOutputValueClass(TableBean.class);

 

              // 5 指定最终输出的数据的kv类型

              job.setOutputKeyClass(TableBean.class);

              job.setOutputValueClass(NullWritable.class);

 

              // 6 指定job的输入原始文件所在目录

              FileInputFormat.setInputPaths(job, new Path(args[0]));

              FileOutputFormat.setOutputPath(job, new Path(args[1]));

 

              // 7 将job中配置的相关参数,以及job所用的java类所在的jar包, 提交给yarn去运行

              boolean result = job.waitForCompletion(true);

              System.exit(result ? 0 : 1);

       }

}


3)运行程序查看结果








1001       小米       1    

1001       小米       1    

1002       华为       2    

1002       华为       2    

1003       格力       3    

1003       格力       3    


缺点:这种方式中,合并的操作是在reduce阶段完成,reduce端的处理压力太大,map节点的运算负载则很低,资源利用率不高,且在reduce阶段极易产生数据倾斜

解决方案: map端实现数据合并


3.4.2 需求2:map端表合并(Distributedcache)

1)分析

适用于关联表中有小表的情形;

可以将小表分发到所有的map节点,这样,map节点就可以在本地对自己所读到的大表数据进行合并并输出最终结果,可以大大提高合并操作的并发度,加快处理速度。

 

 

2)实操案例

(1)先在

推荐阅读
  • 个人学习使用:谨慎参考1Client类importcom.thoughtworks.gauge.Step;importcom.thoughtworks.gauge.T ... [详细]
  • Spring特性实现接口多类的动态调用详解
    本文详细介绍了如何使用Spring特性实现接口多类的动态调用。通过对Spring IoC容器的基础类BeanFactory和ApplicationContext的介绍,以及getBeansOfType方法的应用,解决了在实际工作中遇到的接口及多个实现类的问题。同时,文章还提到了SPI使用的不便之处,并介绍了借助ApplicationContext实现需求的方法。阅读本文,你将了解到Spring特性的实现原理和实际应用方式。 ... [详细]
  • Java太阳系小游戏分析和源码详解
    本文介绍了一个基于Java的太阳系小游戏的分析和源码详解。通过对面向对象的知识的学习和实践,作者实现了太阳系各行星绕太阳转的效果。文章详细介绍了游戏的设计思路和源码结构,包括工具类、常量、图片加载、面板等。通过这个小游戏的制作,读者可以巩固和应用所学的知识,如类的继承、方法的重载与重写、多态和封装等。 ... [详细]
  • 基于Socket的多个客户端之间的聊天功能实现方法
    本文介绍了基于Socket的多个客户端之间实现聊天功能的方法,包括服务器端的实现和客户端的实现。服务器端通过每个用户的输出流向特定用户发送消息,而客户端通过输入流接收消息。同时,还介绍了相关的实体类和Socket的基本概念。 ... [详细]
  • 本文整理了Java面试中常见的问题及相关概念的解析,包括HashMap中为什么重写equals还要重写hashcode、map的分类和常见情况、final关键字的用法、Synchronized和lock的区别、volatile的介绍、Syncronized锁的作用、构造函数和构造函数重载的概念、方法覆盖和方法重载的区别、反射获取和设置对象私有字段的值的方法、通过反射创建对象的方式以及内部类的详解。 ... [详细]
  • vue使用
    关键词: ... [详细]
  • 本文介绍了解决Netty拆包粘包问题的一种方法——使用特殊结束符。在通讯过程中,客户端和服务器协商定义一个特殊的分隔符号,只要没有发送分隔符号,就代表一条数据没有结束。文章还提供了服务端的示例代码。 ... [详细]
  • 在Android开发中,使用Picasso库可以实现对网络图片的等比例缩放。本文介绍了使用Picasso库进行图片缩放的方法,并提供了具体的代码实现。通过获取图片的宽高,计算目标宽度和高度,并创建新图实现等比例缩放。 ... [详细]
  • 向QTextEdit拖放文件的方法及实现步骤
    本文介绍了在使用QTextEdit时如何实现拖放文件的功能,包括相关的方法和实现步骤。通过重写dragEnterEvent和dropEvent函数,并结合QMimeData和QUrl等类,可以轻松实现向QTextEdit拖放文件的功能。详细的代码实现和说明可以参考本文提供的示例代码。 ... [详细]
  • IhaveconfiguredanactionforaremotenotificationwhenitarrivestomyiOsapp.Iwanttwodiff ... [详细]
  • 本文讨论了一个关于cuowu类的问题,作者在使用cuowu类时遇到了错误提示和使用AdjustmentListener的问题。文章提供了16个解决方案,并给出了两个可能导致错误的原因。 ... [详细]
  • 本文介绍了如何使用C#制作Java+Mysql+Tomcat环境安装程序,实现一键式安装。通过将JDK、Mysql、Tomcat三者制作成一个安装包,解决了客户在安装软件时的复杂配置和繁琐问题,便于管理软件版本和系统集成。具体步骤包括配置JDK环境变量和安装Mysql服务,其中使用了MySQL Server 5.5社区版和my.ini文件。安装方法为通过命令行将目录转到mysql的bin目录下,执行mysqld --install MySQL5命令。 ... [详细]
  • ubuntu用sqoop将数据从hive导入mysql时,命令: ... [详细]
  • ***byte(字节)根据长度转成kb(千字节)和mb(兆字节)**parambytes*return*publicstaticStringbytes2kb(longbytes){ ... [详细]
  • 本文介绍了关于Java异常的八大常见问题,包括异常管理的最佳做法、在try块中定义的变量不能用于catch或finally的原因以及为什么Double.parseDouble(null)和Integer.parseInt(null)会抛出不同的异常。同时指出这些问题是由于不同的开发人员开发所导致的,不值得过多思考。 ... [详细]
author-avatar
lily--妹妹
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有