乐山网站建设

专注高端品牌网站建设行业优质供应商！认准小谭建站！
028-86922220135-1821-9792
  • 首页
  • 网站建设
    网站设计包含服务 网站制作流程 有效沟通方式 售后服务
  • 微信开发
  • 案例展示
    乐山小谭建站公司作品案例展示
  • 新闻资讯
  • 解决方案
  • 小谭建站
  • 联系我们

    • 学习日志---hadoop的join处理-创新互联

    Join方法

    创新互联公司主要从事网站制作、成都网站设计、网页设计、企业做网站、公司建网站等业务。立足成都服务梁河,10余年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:028-86922220

    需求：处理input1和input2文件，两个文件中的id都一样，也就是key值一样，value值不同，把两者合并。input1存的是id和名字，input2存的是id和各种信息。

    处理方法一：

    package org.robby.join;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class MyReduceJoin
{
    public static class MapClass extends 
        Mapper
    {
        //map过程需要用到的中间变量
        private Text key = new Text();
        private Text value = new Text();
        private String[] keyValue = null;
        
        @Override
        protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException
        {
            //用逗号分开后传出
            keyValue = value.toString().split(",", 2);
            this.key.set(keyValue[0]);
            this.value.set(keyValue[1]);
            context.write(this.key, this.value);
        }
        
    }
    
    public static class Reduce extends Reducer
    {
        private Text value = new Text();
        
        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException
        {
            StringBuilder valueStr = new StringBuilder();
            
            //reduce过程之所以可以用迭代出相同的id，因为shuffle过程进行了分区，排序，在进入reduce之前，有进行排序和分组，
            //相同的key的值默认分在一组
            for(Text val : values)
            {
                valueStr.append(val);
                valueStr.append(",");
            }
            
            this.value.set(valueStr.deleteCharAt(valueStr.length()-1).toString());
            context.write(key, this.value);
        }
        
    }
    
    public static void main(String[] args) throws Exception
    {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        
        job.setJarByClass(MyReduceJoin.class);
        job.setMapperClass(MapClass.class);
        job.setReducerClass(Reduce.class);
        
        //reduce输出的格式
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        Path outputPath = new Path(args[1]);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(conf).delete(outputPath, true);  

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

    方法一缺点：value值无需，可能第一个文件的value在前，也可能第二个文件的value在前；

    处理方法二：

    引入了一个自定义类型：

    package org.robby.join;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

public class CombineValues implements WritableComparable{
    //这里的自定义类型，实现WritableComparable接口
    //里面的数据使用的是hadoop自带的类型Text
    private Text joinKey;
    private Text flag;
    private Text secondPart;
    
    public void setJoinKey(Text joinKey) {
        this.joinKey = joinKey;
    }
    public void setFlag(Text flag) {
        this.flag = flag;
    }
    public void setSecondPart(Text secondPart) {
        this.secondPart = secondPart;
    }
    public Text getFlag() {
        return flag;
    }
    public Text getSecondPart() {
        return secondPart;
    }
    public Text getJoinKey() {
        return joinKey;
    }
    public CombineValues() {
        //构造时初始化数据，用set添加
        this.joinKey =  new Text();
        this.flag = new Text();
        this.secondPart = new Text();
    }
    
    //序列与反序列化，其中体现为传入文件流，使用hadoop提供的文件流去传送数据                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     
    @Override
    public void write(DataOutput out) throws IOException {
        //因使用的是hadoop自带的Text，因此序列化时，可以用本身的Text，传入流out即可
        this.joinKey.write(out);
        this.flag.write(out);
        this.secondPart.write(out);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
        this.joinKey.readFields(in);
        this.flag.readFields(in);
        this.secondPart.readFields(in);
    }
    @Override
    public int compareTo(CombineValues o) {
        return this.joinKey.compareTo(o.getJoinKey());
    }
    
    @Override
    public String toString() {
        // TODO Auto-generated method stub
        return "[flag="+this.flag.toString()+",joinKey="+this.joinKey.toString()+",secondPart="+this.secondPart.toString()+"]";
    }

}

    处理过程：可以在mapper阶段通过context得到处理的文件是哪一个，因此可以分别处理。

    package org.robby.join;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 
public class MyReduceJoin1
{
    public static class Map extends 
        Mapper
    {
        private CombineValues combineValues = new CombineValues();
        private Text flag = new Text();
        private Text key = new Text();
        private Text value = new Text();
        private String[] keyValue = null;
        
        @Override
        protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException
        {
            //FileSplit是文件块，通过context，文件处理可以的到处理的文件属于哪一个文件
            String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();
            //通过pathName获得处理文件的名字，然后用flag进行标示
            if(pathName.endsWith("input1.txt"))
                flag.set("0");
            else
                flag.set("1");
            
            combineValues.setFlag(flag);
            keyValue = value.toString().split(",", 2);
            combineValues.setJoinKey(new Text(keyValue[0]));
            combineValues.setSecondPart(new Text(keyValue[1]));

            this.key.set(keyValue[0]);
            //将封装的数据传出，key是id，用于分区排序分组，value是自定义的类，在main函数里需要说明
            context.write(this.key, combineValues);
        }
        
    }
    
    public static class Reduce extends Reducer
    {
        private Text value = new Text();
        private Text left = new Text();
        private Text right = new Text();
        
        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException
        {
            //因key一样，因此默认分在一组
            for(CombineValues val : values)
            {
                System.out.println("val:" + val.toString());
                Text secondPar = new Text(val.getSecondPart().toString());
                //根据flag，来判断是左边还是右边
                if(val.getFlag().toString().equals("0")){
                    System.out.println("left :" + secondPar);
                    left.set(secondPar);
                }
                else{
                    System.out.println("right :" + secondPar);
                    right.set(secondPar);
                }
            }
            
            //整合value，输出
            Text output = new Text(left.toString() + "," + right.toString());
            
            context.write(key, output);
        }
        
    }
    
    public static void main(String[] args) throws Exception
    {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        
        job.setJarByClass(MyReduceJoin1.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        
        //这里要指明map的输出，因为默认是Text.class
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CombineValues.class);
        
        //指明reduce的输出
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        //job任务的文件输入和输出形式
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        //job任务的输出与输入文件路径
        Path outputPath = new Path(args[1]);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, outputPath);
        //通个outputPath，查看hdfs是否已有这个文件，有则删除
        outputPath.getFileSystem(conf).delete(outputPath, true);  

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

    缺点：如果两个文件的条数不同，并且还需要把id相同的合并

    处理方法三:

    package org.robby.join;

import java.io.IOException;
import java.util.ArrayList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 
public class MyReduceJoin2
{
    public static class Map extends 
        Mapper
    {
        private CombineValues combineValues = new CombineValues();
        private Text flag = new Text();
        private Text key = new Text();
        private Text value = new Text();
        private String[] keyValue = null;
        
        @Override
        //map的处理和以前一样，分文件加flag标识，用自定义的类型封装输出
        protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException
        {
            String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();
            if(pathName.endsWith("input1.txt"))
                flag.set("0");
            else
                flag.set("1");
            
            combineValues.setFlag(flag);
            keyValue = value.toString().split(",", 2);
            combineValues.setJoinKey(new Text(keyValue[0]));
            combineValues.setSecondPart(new Text(keyValue[1]));

            this.key.set(keyValue[0]);
            context.write(this.key, combineValues);
        }
        
    }
    
    public static class Reduce extends Reducer
    {
        private Text value = new Text();
        private Text left = new Text();
        private ArrayList right = new ArrayList();
        
        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException
        {
            right.clear();
            for(CombineValues val : values)
            {
                //这里id相同的合并，有多个了
                System.out.println("val:" + val.toString());
                Text secondPar = new Text(val.getSecondPart().toString());
                if(val.getFlag().toString().equals("0")){
                    left.set(secondPar);
                }
                else{
                    //文件一是名字，文件二是各种信息，因此存在一个list集合中
                    right.add(secondPar);
                }
            }
            
            for(Text t : right){
                Text output = new Text(left.toString() + "," + t.toString());
                context.write(key, output);
            }
            
        }
        
    }
    
    public static void main(String[] args) throws Exception
    {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        
        job.setJarByClass(MyReduceJoin2.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CombineValues.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        Path outputPath = new Path(args[1]);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(conf).delete(outputPath, true);  

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

    其他处理方法：

    使用distributedCache在mapper环节进行映射；

    主要是重写mapper里面的setup方法，通个context去读取job传入的文件，然后存在mapper对象中，从而使得mapper在每次实现map方法时都可以调用这些预先存入的数据；

    使用setup预先处理input1，则mapper的map方法处理input2即可。

    package org.robby.join;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class MapJoinWithCache {
    public static class Map extends
            Mapper {
        private CombineValues combineValues = new CombineValues();
        private Text flag = new Text();
        private Text key = new Text();
        private Text value = new Text();
        private String[] keyValue = null;
        //这个keyMap就是存文件数据供map共享的
        private HashMap keyMap = null;

        @Override
        //这个map每行都会调用一次，传入数据
        //每次都会访问keyMap集合
        //因为setup方法处理了input1文件，因此这里只需要处理input2就行
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            keyValue = value.toString().split(",", 2);

            String name = keyMap.get(keyValue[0]);
            
            this.key.set(keyValue[0]);
            
            String output = name + "," + keyValue[1];
            this.value.set(output);
            context.write(this.key, this.value);
        }

        @Override
        //这个setup方法是在mapper类初始化运行的方法
        protected void setup(Context context) throws IOException,
                InterruptedException {
            //context传入文件路径
            URI[] localPaths = context.getCacheFiles();
            
            keyMap = new HashMap();
            for(URI url : localPaths){
                 //通过uri打开hdfs文件系统
                 FileSystem fs = FileSystem.get(URI.create("hdfs://hadoop1:9000"), context.getConfiguration());
                 FSDataInputStream in = null;
                 //打开hdfs的对应文件，需要path类创建并传入，获取流对象
                 in = fs.open(new Path(url.getPath()));
                 BufferedReader br=new BufferedReader(new InputStreamReader(in));
                 String s1 = null;
                 while ((s1 = br.readLine()) != null)
                 {
                     keyValue = s1.split(",", 2);
                     
                     keyMap.put(keyValue[0], keyValue[1]);
                     System.out.println(s1);
                 }
                 br.close();
            }
        }
    }

    public static class Reduce extends Reducer {

        //处理都在mpper中进行，reduce迭代分组后的数据就行
        @Override
        protected void reduce(Text key, Iterable values,
                Context context) throws IOException, InterruptedException {
            
            for(Text val : values)
                context.write(key, val);
            
        }

    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(MapJoinWithCache.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

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

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        Path outputPath = new Path(args[1]);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(conf).delete(outputPath, true);

        //其他都一样，这里在job中加入了要传入的文件路径，用作cache
        //可以传入多个文件，文件全路径
        job.addCacheFile(new Path(args[2]).toUri());

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

    其他linux指令：

    [root@hadoop1 dataFile]# wc test*
 6 14 35 test2.txt
 7 16 41 test.txt
13 30 76 total

    可以通过wc查看文件的条数

    附件：http://down.51cto.com/data/2366171

    另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。


    分享题目：学习日志---hadoop的join处理-创新互联
    本文URL：http://lszwz.com/article/jhjci.html

    其他资讯

    售后响应及时

    7×24小时客服热线

    数据备份

    更安全、更高效、更稳定

    价格公道精准

    项目经理精准报价不弄虚作假

    合作无风险

    重合同讲信誉，无效全额退款

    联系我们

    TEL

    135-1821-9792

    028-86922220

    地址：乐山市太升南路288号锦天国际

    快捷导航

    Shortcut

    二维码

    QR CODE

    • 微信公众号

    • 手机端网站

    乐山小谭建站工作室是一家专注从事于高品质视觉体验及互联网设计开发，乐山网站建设，乐山网站设计，乐山网页设计，乐山网站制作，品牌网站建设，营销网站建设，集团网站建设，企业网站建设，外贸网站建设，响应式网站建设，小程序开发，微信开发，企业形象设计，企业宣传视频等服务，小谭建站位于乐山市龙岗区大运软件小镇，小谭建站拥有经验丰富的高级网站建设工程师和一流的网页高端设计人员，具备各种规模与类型网站建设的雄厚实力，在网站建设领域树立了自己独特的设计风格。
    友情链接 交换友情链接
    软文推广公司   梓潼网站建设   成都主机租用   和正源建设   一带一路   四川黔盛装饰   成都棕树机房   网页设计制作   护栏钻孔机   买链接   

    Copyright © 2022 青羊区小谭信息技术咨询服务工作室 乐山建站工作室 All Rights Reserved   蜀ICP备2021004003号-25

    Copyright © 2022 青羊区小谭信息技术咨询服务工作室 乐山建站工作室 

    蜀ICP备2021004003号-25