閱讀482 返回首頁    go 外匯

MapReduce 開發手冊__Hadoop_開發人員指南_E-MapReduce-阿裏雲

在 MapReduce 中使用 OSS

要在 MapReduce 中讀寫 OSS，需要配置如下的參數

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

參數說明：

${accessKeyId}： 您賬號的 AccessKeyId。

${accessKeySecret}： 該 AccessKeyId 對應的密鑰。

${endpoint}： 訪問 OSS 使用的網絡，由您集群所在的 region 決定，當然對應的 OSS 也需要是在集群對應的 region。

具體的值請參考 OSS Endpoint

Word Count

以下示例介紹了如何從 OSS 中讀取文本，然後統計其中單詞的數量。其操作步驟如下：

  1. 程序編寫。以 JAVA 代碼為例，將 Hadoop 官網 WordCount 例子做如下修改。對該實例的修改隻是在代碼中添加了 Access Key ID 和 Access Key Secret 的配置，以便作業有權限訪問 OSS 文件。

    1.  package org.apache.hadoop.examples;
    3.  import java.io.IOException;
    4.  import java.util.StringTokenizer;
    6.  import org.apache.hadoop.conf.Configuration;
    7.  import org.apache.hadoop.fs.Path;
    8.  import org.apache.hadoop.io.IntWritable;
    9.  import org.apache.hadoop.io.Text;
    10.  import org.apache.hadoop.mapreduce.Job;
    11.  import org.apache.hadoop.mapreduce.Mapper;
    12.  import org.apache.hadoop.mapreduce.Reducer;
    13.  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    14.  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    15.  import org.apache.hadoop.util.GenericOptionsParser;
    17.  public class EmrWordCount {
    19.   public static class TokenizerMapper
    20.         extends Mapper<Object, Text, Text, IntWritable>{
    22.      private final static IntWritable one = new IntWritable(1);
    23.      private Text word = new Text();
    25.      public void map(Object key, Text value, Context context
    26.                      ) throws IOException, InterruptedException {
    27.        StringTokenizer itr = new StringTokenizer(value.toString());
    28.        while (itr.hasMoreTokens()) {
    29.          word.set(itr.nextToken());
    30.          context.write(word, one);
    31.        }
    32.      }
    33.    }
    35.    public static class IntSumReducer
    36.         extends Reducer<Text,IntWritable,Text,IntWritable> {
    37.      private IntWritable result = new IntWritable();
    39.      public void reduce(Text key, Iterable<IntWritable> values,
    40.                         Context context
    41.                         ) throws IOException, InterruptedException {
    42.        int sum = 0;
    43.        for (IntWritable val : values) {
    44.          sum += val.get();
    45.        }
    46.        result.set(sum);
    47.        context.write(key, result);
    48.      }
    49.    }
    51.    public static void main(String[] args) throws Exception {
    52.      Configuration conf = new Configuration();
    53.      String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    54.      if (otherArgs.length < 2) {
    55.        System.err.println("Usage: wordcount <in> [<in>...] <out>");
    56.        System.exit(2);
    57.      }
    59.      conf.set("fs.oss.accessKeyId", "${accessKeyId}");
    60.      conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
    61.      conf.set("fs.oss.endpoint","${endpoint}");
    63.      Job job = Job.getInstance(conf, "word count");
    64.      job.setJarByClass(EmrWordCount.class);
    65.      job.setMapperClass(TokenizerMapper.class);
    66.      job.setCombinerClass(IntSumReducer.class);
    67.      job.setReducerClass(IntSumReducer.class);
    68.      job.setOutputKeyClass(Text.class);
    69.      job.setOutputValueClass(IntWritable.class);
    70.      for (int i = 0; i < otherArgs.length - 1; ++i) {
    71.        FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
    72.      }
    73.      FileOutputFormat.setOutputPath(job,
    74.        new Path(otherArgs[otherArgs.length - 1]));
    75.      System.exit(job.waitForCompletion(true) ? 0 : 1);
    76.    }
    77.  }

  2. 編譯程序。首先要將 jdk 和 Hadoop 環境配置好，然後進行如下操作：

    1.  mkdir wordcount_classes
    2.  javac -classpath ${HADOOP_HOME}/share/hadoop/common/hadoop-common-2.6.0.jar:${HADOOP_HOME}/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar:${HADOOP_HOME}/share/hadoop/common/lib/commons-cli-1.2.jar -d wordcount_classes EmrWordCount.java
    3.  jar cvf wordcount.jar -C wordcount_classes .

  3. 創建作業。

    • 將上一步打好的 jar 文件上傳到 OSS，具體可登錄 OSS 官網進行操作。假設 jar 文件在 OSS 上的路徑為 oss://emr/jars/wordcount.jar, 輸入輸出路徑分別為 oss://emr/data/WordCount/Input 和 oss://emr/data/WordCount/Output。

    • 在 E-MapReduce作業 中創建如下作業：

  4. 創建執行計劃。在 E-MapReduce 執行計劃中創建執行計劃，將上一步創建好的作業添加到執行計劃中，策略選擇“立即執行”，這樣 wordcount 作業就會在選定集群中運行起來了。

使用 Maven 工程來管理 MR 作業

當您的工程規模越來越大時，會變得非常複雜，不易管理。我們推薦你采用類似 Maven 這樣的軟件項目管理工具來進行管理。其操作步驟如下：

  1. 安裝 Maven。首先確保您已經安裝了 Maven。

  2. 生成工程框架。在您的工程根目錄處（假設您的工程開發根目錄位置是 D:/workspace）執行如下命令：

    1.  mvn archetype:generate -DgroupId=com.aliyun.emr.hadoop.examples -DartifactId=wordcountv2 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

    mvn 會自動生成一個空的 Sample 工程位於 D:/workspace/wordcountv2（和您指定的 artifactId 一致），裏麵包含一個簡單的 pom.xml 和 App 類（類的包路徑和您指定的 groupId 一致）。

  3. 加入 Hadoop 依賴。使用任意 IDE 打開這個工程，編輯 pom.xml 文件。在 dependencies 內添加如下內容：

    1.      <dependency>
    2.          <groupId>org.apache.hadoop</groupId>
    3.          <artifactId>hadoop-mapreduce-client-common</artifactId>
    4.          <version>2.6.0</version>
    5.      </dependency>
    6.      <dependency>
    7.          <groupId>org.apache.hadoop</groupId>
    8.          <artifactId>hadoop-common</artifactId>
    9.          <version>2.6.0</version>
    10.      </dependency>

  4. 編寫代碼。在 com.aliyun.emr.hadoop.examples 包下和 App 類平行的位置添加新類 WordCount2.java。內容如下：

    1.  package com.aliyun.emr.hadoop.examples;
    3.  import java.io.BufferedReader;
    4.  import java.io.FileReader;
    5.  import java.io.IOException;
    6.  import java.net.URI;
    7.  import java.util.ArrayList;
    8.  import java.util.HashSet;
    9.  import java.util.List;
    10.  import java.util.Set;
    11.  import java.util.StringTokenizer;
    13.  import org.apache.hadoop.conf.Configuration;
    14.  import org.apache.hadoop.fs.Path;
    15.  import org.apache.hadoop.io.IntWritable;
    16.  import org.apache.hadoop.io.Text;
    17.  import org.apache.hadoop.mapreduce.Job;
    18.  import org.apache.hadoop.mapreduce.Mapper;
    19.  import org.apache.hadoop.mapreduce.Reducer;
    20.  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    21.  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    22.  import org.apache.hadoop.mapreduce.Counter;
    23.  import org.apache.hadoop.util.GenericOptionsParser;
    24.  import org.apache.hadoop.util.StringUtils;
    26.  public class WordCount2 {
    28.      public static class TokenizerMapper
    29.              extends Mapper<Object, Text, Text, IntWritable>{
    31.          static enum CountersEnum { INPUT_WORDS }
    33.          private final static IntWritable one = new IntWritable(1);
    34.          private Text word = new Text();
    36.          private boolean caseSensitive;
    37.          private Set<String> patternsToSkip = new HashSet<String>();
    39.          private Configuration conf;
    40.          private BufferedReader fis;
    42.          @Override
    43.          public void setup(Context context) throws IOException,
    44.                  InterruptedException {
    45.              conf = context.getConfiguration();
    46.              caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
    47.              if (conf.getBoolean("wordcount.skip.patterns", true)) {
    48.                  URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
    49.                  for (URI patternsURI : patternsURIs) {
    50.                      Path patternsPath = new Path(patternsURI.getPath());
    51.                      String patternsFileName = patternsPath.getName().toString();
    52.                      parseSkipFile(patternsFileName);
    53.                  }
    54.              }
    55.          }
    57.          private void parseSkipFile(String fileName) {
    58.              try {
    59.                  fis = new BufferedReader(new FileReader(fileName));
    60.                  String pattern = null;
    61.                  while ((pattern = fis.readLine()) != null) {
    62.                      patternsToSkip.add(pattern);
    63.                  }
    64.              } catch (IOException ioe) {
    65.                  System.err.println("Caught exception while parsing the cached file '"
    66.                          + StringUtils.stringifyException(ioe));
    67.              }
    68.          }
    70.          @Override
    71.          public void map(Object key, Text value, Context context
    72.          ) throws IOException, InterruptedException {
    73.              String line = (caseSensitive) ?
    74.                      value.toString() : value.toString().toLowerCase();
    75.              for (String pattern : patternsToSkip) {
    76.                  line = line.replaceAll(pattern, "");
    77.              }
    78.              StringTokenizer itr = new StringTokenizer(line);
    79.              while (itr.hasMoreTokens()) {
    80.                  word.set(itr.nextToken());
    81.                  context.write(word, one);
    82.                  Counter counter = context.getCounter(CountersEnum.class.getName(),
    83.                          CountersEnum.INPUT_WORDS.toString());
    84.                  counter.increment(1);
    85.              }
    86.          }
    87.      }
    89.      public static class IntSumReducer
    90.              extends Reducer<Text,IntWritable,Text,IntWritable> {
    91.          private IntWritable result = new IntWritable();
    93.          public void reduce(Text key, Iterable<IntWritable> values,
    94.                             Context context
    95.          ) throws IOException, InterruptedException {
    96.              int sum = 0;
    97.              for (IntWritable val : values) {
    98.                  sum += val.get();
    99.              }
    100.              result.set(sum);
    101.              context.write(key, result);
    102.          }
    103.      }
    105.      public static void main(String[] args) throws Exception {
    106.          Configuration conf = new Configuration();
    108.          conf.set("fs.oss.accessKeyId", "${accessKeyId}");
    109.         conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
    110.          conf.set("fs.oss.endpoint","${endpoint}");
    112.          GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    113.          String[] remainingArgs = optionParser.getRemainingArgs();
    114.          if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
    115.              System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
    116.              System.exit(2);
    117.          }
    118.          Job job = Job.getInstance(conf, "word count");
    119.          job.setJarByClass(WordCount2.class);
    120.          job.setMapperClass(TokenizerMapper.class);
    121.          job.setCombinerClass(IntSumReducer.class);
    122.          job.setReducerClass(IntSumReducer.class);
    123.          job.setOutputKeyClass(Text.class);
    124.          job.setOutputValueClass(IntWritable.class);
    126.          List<String> otherArgs = new ArrayList<String>();
    127.          for (int i=0; i < remainingArgs.length; ++i) {
    128.              if ("-skip".equals(remainingArgs[i])) {
    129.                  job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
    130.                  job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
    131.              } else {
    132.                  otherArgs.add(remainingArgs[i]);
    133.              }
    134.          }
    135.          FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
    136.          FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
    138.          System.exit(job.waitForCompletion(true) ? 0 : 1);
    139.      }
    140.  }

    其中的 EMapReduceOSSUtil 類代碼請參見如下示例，放在和 WordCount2 相同目錄：

    1.  package com.aliyun.emr.hadoop.examples;
    3.  import org.apache.hadoop.conf.Configuration;
    5.  public class EMapReduceOSSUtil {
    7.      private static String SCHEMA = "oss://";
    8.      private static String AKSEP = ":";
    9.      private static String BKTSEP = "@";
    10.      private static String EPSEP = ".";
    11.      private static String HTTP_HEADER = "https://";
    13.      /**
    14.       * complete OSS uri
    15.       * convert uri like: oss://bucket/path  to  oss://accessKeyId:accessKeySecret@bucket.endpoint/path
    16.       * ossref do not need this
    17.       *
    18.       * @param oriUri original OSS uri
    19.       */
    20.      public static String buildOSSCompleteUri(String oriUri, String akId, String akSecret, String endpoint) {
    21.          if (akId == null) {
    22.              System.err.println("miss accessKeyId");
    23.              return oriUri;
    24.          }
    25.          if (akSecret == null) {
    26.              System.err.println("miss accessKeySecret");
    27.              return oriUri;
    28.          }
    29.          if (endpoint == null) {
    30.              System.err.println("miss endpoint");
    31.              return oriUri;
    32.          }
    34.          int index = oriUri.indexOf(SCHEMA);
    35.          if (index == -1 || index != 0) {
    36.              return oriUri;
    37.          }
    39.          int bucketIndex = index + SCHEMA.length();
    40.          int pathIndex = oriUri.indexOf("/", bucketIndex);
    41.          String bucket = null;
    42.          if (pathIndex == -1) {
    43.              bucket = oriUri.substring(bucketIndex);
    44.          } else {
    45.              bucket = oriUri.substring(bucketIndex, pathIndex);
    46.          }
    48.          StringBuilder retUri = new StringBuilder();
    49.          retUri.append(SCHEMA)
    50.                  .append(akId)
    51.                  .append(AKSEP)
    52.                  .append(akSecret)
    53.                  .append(BKTSEP)
    54.                  .append(bucket)
    55.                  .append(EPSEP)
    56.                  .append(stripHttp(endpoint));
    58.          if (pathIndex > 0) {
    59.              retUri.append(oriUri.substring(pathIndex));
    60.          }
    62.          return retUri.toString();
    63.      }
    65.      public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
    66.          return buildOSSCompleteUri(oriUri, conf.get("fs.oss.accessKeyId"), conf.get("fs.oss.accessKeySecret"), conf.get("fs.oss.endpoint"));
    67.      }
    69.      private static String stripHttp(String endpoint) {
    70.          if (endpoint.startsWith(HTTP_HEADER)) {
    71.              return endpoint.substring(HTTP_HEADER.length());
    72.          }
    73.          return endpoint;
    74.      }
    75.  }

  5. 編譯並打包上傳。在工程的目錄下，執行如下命令：

    1.  mvn clean package -DskipTests

    您即可在工程目錄的 target 目錄下看到一個 wordcountv2-1.0-SNAPSHOT.jar，這個就是作業 jar 包了。請您將這個 jar 包上傳到 OSS 中。

  6. 創建作業。在 E-MapReduce 中新建一個作業，請使用類似如下的參數配置：

    1.  jar ossref://yourBucket/yourPath/wordcountv2-1.0-SNAPSHOT.jar com.aliyun.emr.hadoop.examples.WordCount2 -Dwordcount.case.sensitive=true oss://yourBucket/yourPath/The_Sorrows_of_Young_Werther.txt oss://yourBucket/yourPath/output -skip oss://yourBucket/yourPath/patterns.txt

    這裏的 yourBucket 是您的一個 OSS bucket，yourPath 是這個 bucket 上的一個路徑，需要您按照實際情況填寫。請您將oss://yourBucket/yourPath/The_Sorrows_of_Young_Werther.txt和oss://yourBucket/yourPath/patterns.txt這兩個用來處理相關資源的文件下載下來並放到您的 OSS 上。作業需要資源可以從下麵下載，然後放到您的 OSS 對應目錄下。

    資源下載：The_Sorrows_of_Young_Werther.txtpatterns.txt

  7. 創建執行計劃並運行。在 E-MapReduce 中創建執行計劃，關聯這個作業並運行。

最後更新：2016-11-23 16:03:59

  上一篇：go SDK-Release__Spark_開發人員指南_E-MapReduce-阿裏雲
  下一篇：go Hive 開發手冊__Hadoop_開發人員指南_E-MapReduce-阿裏雲