[TOC] # 分析 適用于關聯表中有小表的情形 可以將小表分發到所有的map節點,這樣,map節點就可以在本地對自己所讀到的大表數據進行合并并輸出最終結果,可以大大提高合并操作的并發速度 **DistributedCacheDriver緩存文件** 1. 加快緩存數據 ~~~ job.addCacheFile(new URI("file://e:/cache/od.txt")); ~~~ 2. map端join的邏輯不需要reduce階段,設置reduceTask數量為0 ~~~ job.setNumReduceTask(0); ~~~ **讀取緩存的文件數據** setup方法中 1. 獲取緩存的文件 2. 循環讀取緩存文件一行 3. 切割 4. 緩存數據到集合 5. 關流 map方法中 1. 獲取一行 2. 截取 3. 獲取訂單id 4. 獲取商品名稱 5. 拼接 6. 寫出 # 代碼 延續上面reducer端合并的需求 ## map類 ~~~ import org.apache.commons.lang.StringUtils; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.BufferedReader; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStreamReader; import java.util.HashMap; import java.util.Map; public class DistributedCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> { Map<String, String> pdMap = new HashMap<>(); Text k = new Text(); @Override protected void setup(Context context) throws IOException, InterruptedException { //獲取緩存的文件 BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream("pd.txt"), "UTF-8")); String line; while (StringUtils.isNotEmpty(line = reader.readLine())) { //切割 String[] fields = line.split(" "); //緩存數據到集合 pdMap.put(fields[0], fields[1]); } } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //獲取一行 String line = value.toString(); //截取 String[] fields = line.split(" "); //獲取產品id String pId = fields[1]; //獲取商品名稱 String pdName = pdMap.get(pId); //拼接 k.set(line + "\t" + pdName); //寫出 context.write(k, NullWritable.get()); } } ~~~ ## 驅動類 ~~~ 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; import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; public class TableDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException { //job信息 Configuration conf = new Configuration(); Job job = Job.getInstance(conf); //設置加載jar包路徑 job.setJarByClass(TableDriver.class); //關聯map job.setMapperClass(DistributedCacheMapper.class); //設置最終的輸出類型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); //設置輸入文件和輸出路徑 FileInputFormat.setInputPaths(job, new Path("/Users/jdxia/Desktop/website/data/input/order.txt")); FileOutputFormat.setOutputPath(job, new Path("/Users/jdxia/Desktop/website/data/output")); //加載緩存數據 job.addCacheFile(new URI("file:///Users/jdxia/Desktop/website/data/input/pd.txt")); //不需要reduce job.setNumReduceTasks(0); job.waitForCompletion(true); } } ~~~