すべてのプロダクト
Search

このプロダクト

    E-MapReduce:MapReduce ジョブの開発

    ドキュメントセンター

    E-MapReduce:MapReduce ジョブの開発

    最終更新日:Apr 01, 2025

    このトピックでは、V3.27.0 の E-MapReduce（EMR）クラスターを例として使用して、EMR クラスターで MapReduce ジョブを開発する方法を示します。

    MapReduce ジョブを使用した OSS データの処理

    MapReduce ジョブを使用してオブジェクトストレージサービス（OSS）からデータを読み取るか、OSS にデータを書き込むには、次のパラメーターを設定する必要があります。

    説明

    ALIBABA_CLOUD_ACCESS_KEY_ID および ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。詳細については、「アクセスクレデンシャルの管理」をご参照ください。

    conf.set("fs.oss.accessKeyId", "${accessKeyId}");
// アクセスキーIDを設定します。
conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
// アクセスキーシークレットを設定します。
conf.set("fs.oss.endpoint","${endpoint}");
// エンドポイントを設定します。

    パラメーターの説明：

    • ${accessKeyId}: Alibaba Cloud アカウントの AccessKey ID。

    • ${accessKeySecret}: Alibaba Cloud アカウントの AccessKey シークレット。

    • ${endpoint}: OSS のエンドポイント。

      このパラメーターを、EMR クラスターが存在するリージョンに対応するエンドポイントに設定します。使用する OSS バケットが EMR クラスターと同じリージョンにあることを確認する必要があります。リージョンとエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。

    WordCount の例

    この例では、MapReduce ジョブを使用して OSS からテキストを読み取り、テキスト内の各単語の出現回数をカウントし、結果を OSS バケットに書き戻す方法を示します。

    1. SSH モードでクラスターにログオンします。詳細については、「クラスターにログオンする」をご参照ください。

    2. 次のコマンドを実行して、wordcount_classes という名前のディレクトリを作成します。

      mkdir wordcount_classes

    3. EmrWordCount.java という名前のファイルを作成します。

      1. 次のコマンドを実行して、EmrWordCount.java という名前のファイルを作成し、ファイルを開きます。

        vim EmrWordCount.java

      2. I キーを押して、編集モードに切り替えます。

      3. 次の情報を EmrWordCount.java ファイルに追加します。

        package org.apache.hadoop.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}");
// アクセスキーIDを設定します。
     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);
   }
 }

      4. Esc キーを押して、編集モードを終了します。次に、:wq と入力して、ファイルを保存して閉じます。

    4. 作成したファイルをコンパイルしてパッケージ化します。

      1. 次のコマンドを実行して、プログラムをコンパイルします。

        javac -classpath <HADOOP_HOME>/share/hadoop/common/hadoop-common-X.X.X.jar:<HADOOP_HOME>/share/hadoop/mapreduce/hadoop-mapreduce-client-core-X.X.X.jar:<HADOOP_HOME>/share/hadoop/common/lib/commons-cli-1.2.jar -d wordcount_classes EmrWordCount.java

        • HADOOP_HOME: Hadoop のインストールディレクトリ。ほとんどの場合、/usr/lib/hadoop-current ディレクトリが使用されます。

          env |grep hadoop コマンドを実行して、インストールディレクトリを取得できます。

        • X.X.X: JAR パッケージのバージョン。クラスター内の Hadoop のバージョンと同じである必要があります。

          hadoop-common-X.X.X.jar ファイルの場合は、<HADOOP_HOME>/share/hadoop/common/ ディレクトリでバージョンを確認できます。 hadoop-mapreduce-client-core-X.X.X.jar ファイルの場合は、<HADOOP_HOME>/share/hadoop/mapreduce/ ディレクトリでバージョンを確認できます。

      2. 次のコマンドを実行して、コンパイルされたプログラムを JAR ファイルにパッケージ化します。

        jar cvf wordcount.jar -C wordcount_classes .
        説明

        この例では、JAR ファイルは wordcount.jar で、デフォルトディレクトリ /root に保存されます。

    5. ジョブを作成します。

      1. wordcount.jar手順 4 で取得した オブジェクトのアップロード ファイルを OSS にアップロードします。詳細については、「」をご参照ください。

        この例では、ファイルは oss://<yourBucketName>/jars/wordcount.jar にアップロードされます。

      2. EMR コンソールで MapReduce ジョブを作成します。詳細については、「Hadoop MapReduce ジョブの設定」をご参照ください。

        ジョブの内容：

        ossref://<yourBucketName>/jars/wordcount.jar org.apache.hadoop.examples.EmrWordCount oss://<yourBucketName>/data/WordCount/Input oss://<yourBucketName>/data/WordCount/Output

        コード内の <yourBucketName> を、使用する OSS バケットの名前に置き換えます。 oss://<yourBucketName>/data/WordCount/Input は入力パスを、oss://<yourBucketName>/data/WordCount/Output は出力パスを示します。

      3. [実行] ボタンをクリックします。

        MapReduce ジョブがクラスターで実行を開始します。

    Wordcount2 の例

    プロジェクトが大規模な場合は、Maven などのツールを使用してプロジェクト内のジョブを管理できます。この例では、Maven を使用して MapReduce ジョブを管理する方法を示します。

    1. オンプレミスマシンに Maven と Java をインストールします。

      この例では、Maven 3.0 と Java 1.8 を使用します。

    2. 次のコマンドを実行して、プロジェクトフレームワークを生成します。

      この例では、プロジェクトを開発するためのルートディレクトリは D:/workspace です。

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

      コマンドを実行すると、指定された artifactId と一致する D:/workspace/wordcountv2 ディレクトリに空のサンプルプロジェクトが自動的に生成されます。プロジェクトには、pom.xml という名前のファイルと App という名前のクラスが含まれています。App クラスのパッケージパスは、指定された groupId と一致しています。

    3. Hadoop の依存関係を追加します。

      統合開発環境（IDE）を使用してサンプルプロジェクトを開き、pom.xml ファイルを編集します。Hadoop 2.8.5 を使用する場合は、次の内容をファイルに追加します。

      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-mapreduce-client-common</artifactId>
         <version>2.8.5</version>
</dependency>
<dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-common</artifactId>
         <version>2.8.5</version>
</dependency>

    4. コードを記述します。

      1. EMapReduceOSSUtil という名前の新しいクラスを、com.aliyun.emr.hadoop.examples パッケージ内の App クラスと同じディレクトリ レベルに追加します。

        package com.aliyun.emr.hadoop.examples;
 import org.apache.hadoop.conf.Configuration;
 public class EMapReduceOSSUtil {
     private static String SCHEMA = "oss://";
     private static String EPSEP = ".";
     private static String HTTP_HEADER = "http://";
     /**
      * OSS URIを完成させます
      * oss://bucket/path のような URI を oss://bucket.endpoint/path に変換します
      * ossref はこれを必要としません
      *
      * @param oriUri 元の OSS URI
      */
     public static String buildOSSCompleteUri(String oriUri, String endpoint) {
         if (endpoint == null) {
             System.err.println("エンドポイントがありません");
             return oriUri;
         }
         int index = oriUri.indexOf(SCHEMA);
         if (index == -1 || index != 0) {
             return oriUri;
         }
         int bucketIndex = index + SCHEMA.length();
         int pathIndex = oriUri.indexOf("/", bucketIndex);
         String bucket = null;
         if (pathIndex == -1) {
             bucket = oriUri.substring(bucketIndex);
         } else {
             bucket = oriUri.substring(bucketIndex, pathIndex);
         }
         StringBuilder retUri = new StringBuilder();
         retUri.append(SCHEMA)
                 .append(bucket)
                 .append(EPSEP)
                 .append(stripHttp(endpoint));
         if (pathIndex > 0) {
             retUri.append(oriUri.substring(pathIndex));
         }
         return retUri.toString();
     }
     public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
         return buildOSSCompleteUri(oriUri, conf.get("fs.oss.endpoint"));
     }
     private static String stripHttp(String endpoint) {
         if (endpoint.startsWith(HTTP_HEADER)) {
             return endpoint.substring(HTTP_HEADER.length());
         }
         return endpoint;
     }
 }

      2. WordCount2.java という名前の新しいクラスを、com.aliyun.emr.hadoop.examples パッケージ内の App クラスと同じディレクトリレベルに追加します。

        package com.aliyun.emr.hadoop.examples;
 import java.io.BufferedReader;
 import java.io.FileReader;
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 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.mapreduce.Counter;
 import org.apache.hadoop.util.GenericOptionsParser;
 import org.apache.hadoop.util.StringUtils;
 public class WordCount2 {
     public static class TokenizerMapper
             extends Mapper<Object, Text, Text, IntWritable>{
         static enum CountersEnum { INPUT_WORDS }
         private final static IntWritable one = new IntWritable(1);
         private Text word = new Text();
         private boolean caseSensitive;
         private Set<String> patternsToSkip = new HashSet<String>();
         private Configuration conf;
         private BufferedReader fis;
         @Override
         public void setup(Context context) throws IOException,
                 InterruptedException {
             conf = context.getConfiguration();
             caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
             if (conf.getBoolean("wordcount.skip.patterns", true)) {
                 URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
                 for (URI patternsURI : patternsURIs) {
                     Path patternsPath = new Path(patternsURI.getPath());
                     String patternsFileName = patternsPath.getName().toString();
                     parseSkipFile(patternsFileName);
                 }
             }
         }
         private void parseSkipFile(String fileName) {
             try {
                 fis = new BufferedReader(new FileReader(fileName));
                 String pattern = null;
                 while ((pattern = fis.readLine()) != null) {
                     patternsToSkip.add(pattern);
                 }
             } catch (IOException ioe) {
                 System.err.println("キャッシュされたファイル '" + StringUtils.stringifyException(ioe) + "' を解析しているときに例外が発生しました");
             }
         }
         @Override
         public void map(Object key, Text value, Context context
         ) throws IOException, InterruptedException {
             String line = (caseSensitive) ?
                     value.toString() : value.toString().toLowerCase();
             for (String pattern : patternsToSkip) {
                 line = line.replaceAll(pattern, "");
             }
             StringTokenizer itr = new StringTokenizer(line);
             while (itr.hasMoreTokens()) {
                 word.set(itr.nextToken());
                 context.write(word, one);
                 Counter counter = context.getCounter(CountersEnum.class.getName(),
                         CountersEnum.INPUT_WORDS.toString());
                 counter.increment(1);
             }
         }
     }
     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();
         conf.set("fs.oss.accessKeyId", "${accessKeyId}");
// アクセスキーIDを設定します。
         conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
// アクセスキーシークレットを設定します。
         conf.set("fs.oss.endpoint","${endpoint}");
// エンドポイントを設定します。
         GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
         String[] remainingArgs = optionParser.getRemainingArgs();
         if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
             System.err.println("使用方法: wordcount <in> <out> [-skip skipPatternFile]");
             System.exit(2);
         }
         Job job = Job.getInstance(conf, "word count");
         job.setJarByClass(WordCount2.class);
         job.setMapperClass(TokenizerMapper.class);
         job.setCombinerClass(IntSumReducer.class);
         job.setReducerClass(IntSumReducer.class);
         job.setOutputKeyClass(Text.class);
         job.setOutputValueClass(IntWritable.class);
         List<String> otherArgs = new ArrayList<String>();
         for (int i=0; i < remainingArgs.length; ++i) {
             if ("-skip".equals(remainingArgs[i])) {
                 job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
                 job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
             } else {
                 otherArgs.add(remainingArgs[i]);
             }
         }
         FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
         FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
         System.exit(job.waitForCompletion(true) ? 0 : 1);
     }
 }

    5. プロジェクトのルートディレクトリで、次のコマンドを実行して、2つの新しいクラスのファイルをコンパイルおよびパッケージ化します。

      mvn clean package -DskipTests

      wordcountv2-1.0-SNAPSHOT.jar という名前の JAR ファイルが、プロジェクトの target ディレクトリに生成されます。

    6. ジョブを作成します。

      1. wordcountv2-1.0-SNAPSHOT.jar手順 5 で取得した オブジェクトのアップロード ファイルを OSS にアップロードします。詳細については、「」をご参照ください。

        この例では、ファイルは oss://<yourBucketName>/jars/wordcountv2-1.0-SNAPSHOT.jar にアップロードされます。

      2. 次のファイルをダウンロードして、OSS ディレクトリにアップロードします。

        説明

        The_Sorrows_of_Young_Werther.txt は、各単語の出現回数をカウントする必要があるテキストファイルです。 patterns.txt には、無視する単語パターンがリストされています。

      3. EMR コンソールで MapReduce ジョブを作成します。詳細については、「Hadoop MapReduce ジョブの設定」をご参照ください。

        ジョブの内容：

        ossref://<yourBucketName>/jars/wordcountv2-1.0-SNAPSHOT.jar com.aliyun.emr.hadoop.examples.WordCount2 -D wordcount.case.sensitive=true oss://<yourBucketName>/jars/The_Sorrows_of_Young_Werther.txt oss://<yourBucketName>/jars/output -skip oss://<yourBucketName>/jars/patterns.txt

        コード内の <yourBucketName> を、使用する OSS バケットの名前に置き換えます。 oss://<yourBucketName>/jars/output は出力パスを示します。

      4. [実行] ボタンをクリックします。

        MapReduce ジョブがクラスターで実行を開始します。