共計 3115 個字符，預計需要花費 8 分鐘才能閱讀完成。

這篇文章主要介紹“MapReduce Map Join 怎么使用”，在日常操作中，相信很多人在 MapReduce Map Join 怎么使用問題上存在疑惑，丸趣 TV 小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”MapReduce Map Join 怎么使用”的疑惑有所幫助！接下來，請跟著丸趣 TV 小編一起來學習吧！

1. 樣例數據  
 

011990-99999 SIHCCAJAVRI
012650-99999 TYNSET-HANSMOEN

012650-99999 194903241200 111
012650-99999 194903241800 78
011990-99999 195005150700 0
011990-99999 195005151200 22
011990-99999 195005151800 -11

2. 需求  
 

3. 思路、代碼  
將足夠小的關聯文件（即氣象臺信息）添加到分布式緩存，然后在每個 Mapper 端讀取被緩存到本地的全量氣象臺信息，再與天氣信息相關聯。

import org.apache.hadoop.conf.Configuration;
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.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

 static class RecordMapper extends Mapper LongWritable, Text, Text, Text  {
 private Map String, String  stationMap = new HashMap String, String 
 @Override
 protected void setup(Context context) throws IOException, InterruptedException {
 // 預處理，把要關聯的文件加載到緩存中
 Path[] paths = context.getLocalCacheFiles();
 // 新的檢索緩存文件的 API 是  context.getCacheFiles() ，而  context.getLocalCacheFiles()  被棄用
 // 然而  context.getCacheFiles()  返回的是  HDFS  路徑； context.getLocalCacheFiles()  返回的才是本地路徑
 // 這里只緩存了一個文件，所以取第一個即可
 BufferedReader reader = new BufferedReader(new FileReader(paths[0].toString()));
 String line = null;
 try { while ((line = reader.readLine()) != null) { String[] vals = line.split( \\t 
 if (vals.length == 2) { stationMap.put(vals[0], vals[1]);
 }
 }
 } catch (Exception e) { e.printStackTrace();
 } finally { reader.close();
 }
 super.setup(context);
 }
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] vals = value.toString().split( \\t 
 if (vals.length == 3) { String stationName = stationMap.get(vals[0]); //Join
 stationName = stationName == null ?   : stationName;
 context.write(new Text(vals[0]),
 new Text(stationName +  \t  + vals[1] +  \t  + vals[2]));
 }
 }
 }
 public static void main(String[] args) throws Exception { Configuration conf = new Configuration();
 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
 if (otherArgs.length != 3) {
 System.err.println( Parameter number is wrong, please enter three parameters：ncdc input   station input   output 
 System.exit(-1);
 }
 Path inputPath = new Path(otherArgs[0]);
 Path stationPath = new Path(otherArgs[1]);
 Path outputPath = new Path(otherArgs[2]);
 Job job = Job.getInstance(conf,  MapJoin 
 job.setJarByClass(MapJoin.class);
 FileInputFormat.addInputPath(job, inputPath);
 FileOutputFormat.setOutputPath(job, outputPath);
 job.addCacheFile(stationPath.toUri()); // 添加緩存文件，可添加多個
 job.setMapperClass(RecordMapper.class);
 job.setMapOutputKeyClass(Text.class);
 System.exit(job.waitForCompletion(true) ? 0 : 1);
 }
}

4. 運行結果

到此，關于“MapReduce Map Join 怎么使用”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注丸趣 TV 網站，丸趣 TV 小編會繼續努力為大家帶來更多實用的文章！