hadoop_MapReduce map端join算法實作

Join算法

• • • • 原理闡述
      • 實作示例

原理闡述

• 适用于關聯表中有小表的情形；
• 可以将小表分發到所有的map節點，這樣，map節點就可以在本地對自己所讀到的大表資料進行join并輸出最終結果，可以大大提高join操作的并發度，加快處理速度

實作示例

map端的初始化方法當中擷取緩存檔案：

package MapJoin;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class MapJoinMap extends Mapper<LongWritable, Text,Text,Text> {

    //擷取緩存中的資料
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //在這裡讀取緩存檔案
        //獲得緩存的檔案
        URI[] cacheFiles = DistributedCache.getCacheFiles(context.getConfiguration());
        //cacheFiles[0]
        //通過URI 獲得檔案系統
        FileSystem fileSystem = FileSystem.get(cacheFiles[0], context.getConfiguration());
        //通過檔案系統打開緩存的檔案
        FSDataInputStream inputStream = fileSystem.open(new Path(cacheFiles[0]));
        //使用BufferedReader讀取資料
        BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(inputStream,"utf-8"));
        String line="";
        while ((line=bufferedReader.readLine())!=null){
            
        }
        bufferedReader.close();
        fileSystem.close();

    }

    //處理超大量資料的代碼    10億
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

    }
}

           

定義程式運作main方法：

package MapJoin;

import WC.WordCountDriver;
import WC.WordCountMap;
import WC.WordCountReduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.net.URI;

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

        Configuration conf = new Configuration();

        //設定叢集緩存的資料
        //這個資料必須在HDFS
        DistributedCache.addCacheFile(new URI("hdfs://192.168.10.131:8020/orders.txt"), conf);

        Job job = Job.getInstance(conf, "MapJoin");

        //設定程式的主類
        job.setJarByClass(Driver.class);

        //設定Map程式代碼
        job.setMapperClass(MapJoinMap.class);

        //設定Map輸出的key  value的類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //設定去哪裡讀取資料
        FileInputFormat.addInputPath(job, new Path(""));
        //設定最終結果寫到哪裡去
        FileOutputFormat.setOutputPath(job, new Path(""));

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);

    }
}