DataWorks：EMR MR節點

準備初始資料

建立樣本檔案input01.txt檔案內容如下。

hadoop emr hadoop dw
hive hadoop
dw emr

上傳初始資料檔案

1. 登入OSS管理主控台，單擊左側導覽列的Bucket列表。

2. 單擊目標Bucket名稱，進入檔案管理頁面。

   本文樣本使用的Bucket為onaliyun-bucket-2。

3. 單擊建立目錄，建立初始資料及JAR資源的存放目錄。

   • 配置目錄名為emr/datas/wordcount02/inputs，建立初始資料的存放目錄。

   • 配置目錄名為emr/jars，建立JAR資源的存放目錄。

4. 上傳初始資料檔案至初始資料的存放目錄。

   • 進入/emr/datas/wordcount02/inputs路徑，單擊上傳檔案。

   • 在待上傳檔案地區單擊掃描檔案，添加input01.txt檔案至Bucket，單擊上傳檔案。

使用MapReduce讀取OSS檔案並產生JAR包

1. 開啟已建立的IDEA專案，添加pom依賴。

           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-mapreduce-client-common</artifactId>
               <version>2.8.5</version> <!--因為EMR-MR用的是2.8.5-->
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>2.8.5</version>
           </dependency>

2. 在MapReduce中讀寫OSS檔案，需要配置如下參數。

   重要

   風險提示： 阿里雲帳號AccessKey擁有所有API的存取權限，建議您使用RAM使用者進行API訪問或日常營運。強烈建議不要將AccessKey ID和AccessKey Secret儲存到工程代碼裡或者任何容易被泄露的地方，AccessKey泄露會威脅您帳號下所有資源的安全。以下程式碼範例僅供參考，請妥善保管好您的AccessKey資訊。

   conf.set("fs.oss.accessKeyId", "${accessKeyId}");
   conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
   conf.set("fs.oss.endpoint","${endpoint}");

   參數說明如下：

   • ${accessKeyId}：阿里雲帳號的AccessKey ID。

   • ${accessKeySecret}：阿里雲帳號的AccessKey Secret。

   • ${endpoint}：OSS對外服務的訪問網域名稱。由您叢集所在的地區決定，對應的OSS也需要是在叢集對應的地區，詳情請參見OSS地區和訪問網域名稱。

   以Java代碼為例，修改Hadoop官網WordCount樣本，即在代碼中添加AccessKey ID和AccessKey Secret的配置，以便作業有許可權訪問OSS檔案。

   範例程式碼

   package cn.apache.hadoop.onaliyun.examples;
   
   import java.io.IOException;
   import java.util.StringTokenizer;
   
   import org.apache.hadoop.conf.Configuration;
   import org.apache.hadoop.fs.Path;
   import org.apache.hadoop.io.IntWritable;
   import org.apache.hadoop.io.Text;
   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.output.FileOutputFormat;
   import org.apache.hadoop.util.GenericOptionsParser;
   
   public class EmrWordCount {
       public static class TokenizerMapper
               extends Mapper<Object, Text, Text, IntWritable> {
           private final static IntWritable one = new IntWritable(1);
           private Text word = new Text();
   
           public void map(Object key, Text value, Context context
           ) throws IOException, InterruptedException {
               StringTokenizer itr = new StringTokenizer(value.toString());
               while (itr.hasMoreTokens()) {
                   word.set(itr.nextToken());
                   context.write(word, one);
               }
           }
       }
   
       public static class IntSumReducer
               extends Reducer<Text, IntWritable, Text, IntWritable> {
           private IntWritable result = new IntWritable();
   
           public void reduce(Text key, Iterable<IntWritable> values,
                              Context context
           ) throws IOException, InterruptedException {
               int sum = 0;
               for (IntWritable val : values) {
                   sum += val.get();
               }
               result.set(sum);
               context.write(key, result);
           }
       }
   
       public static void main(String[] args) throws Exception {
           Configuration conf = new Configuration();
           String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
           if (otherArgs.length < 2) {
               System.err.println("Usage: wordcount <in> [<in>...] <out>");
               System.exit(2);
           }
           conf.set("fs.oss.accessKeyId", "${accessKeyId}"); // 
           conf.set("fs.oss.accessKeySecret", "${accessKeySecret}"); // 
           conf.set("fs.oss.endpoint", "${endpoint}"); //
           Job job = Job.getInstance(conf, "word count");
           job.setJarByClass(EmrWordCount.class);
           job.setMapperClass(TokenizerMapper.class);
           job.setCombinerClass(IntSumReducer.class);
           job.setReducerClass(IntSumReducer.class);
           job.setOutputKeyClass(Text.class);
           job.setOutputValueClass(IntWritable.class);
           for (int i = 0; i < otherArgs.length - 1; ++i) {
               FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
           }
           FileOutputFormat.setOutputPath(job,
                   new Path(otherArgs[otherArgs.length - 1]));
           System.exit(job.waitForCompletion(true) ? 0 : 1);
       }
   }
                                   

3. 編輯完上述Java代碼後將該代碼產生JAR包。樣本產生的JAR包為onaliyun_mr_wordcount-1.0-SNAPSHOT.jar。