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);
}
}