DataWorks:EMR MR ノードの作成

初期データを準備する

input01.txt という名前で、次の初期データを含むファイルを作成します。

hadoop emr hadoop dw
hive hadoop
dw emr

初期データを格納するファイルをアップロードする

1. OSS コンソール にログインします。 左側のナビゲーションウィンドウで、[バケット] をクリックします。

2. バケットページで、目的のバケットを見つけ、バケット名をクリックして [オブジェクト] ページに移動します。

   この例では、onaliyun-bucket-2 バケットを使用します。

3. [ディレクトリの作成] をクリックして、初期データと JAR リソースを格納するために使用するディレクトリを作成します。

   • [ディレクトリ名] を emr/datas/wordcount02/inputs に設定して、初期データを格納するために使用するディレクトリを作成します。

   • [ディレクトリ名] を emr/jars に設定して、JAR リソースを格納するために使用するディレクトリを作成します。

4. 初期データを格納するファイルを emr/datas/wordcount02/inputs ディレクトリにアップロードします。

   • /emr/datas/wordcount02/inputs ディレクトリに移動し、[オブジェクトのアップロード] をクリックします。

   • [アップロードするファイル] セクションで、[ファイルを選択] をクリックし、input01.txt ファイルをバケットにアップロードします。

EMR MR ノードを使用して OSS オブジェクトを読み取り、JAR パッケージを生成する

1. 既存の IntelliJ IDEA プロジェクトを開き、Project Object Model (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. OSS オブジェクトからデータを読み書きするために、次のパラメータを構成します。

   重要

   Alibaba Cloud アカウントには、すべての API オペレーションを呼び出す権限があります。 Alibaba Cloud アカウントの AccessKey ペアが漏洩した場合、Alibaba Cloud アカウント内のすべてのリソースが重大なセキュリティリスクにさらされる可能性があります。 Alibaba Cloud アカウントの AccessKey ID と AccessKey シークレットをプロジェクトコードまたは簡単に見つけられる場所に保存しないことをお勧めします。 API オペレーションの呼び出しや日常の O&M の実行には、RAM ユーザーを使用することをお勧めします。 次のサンプルコードは参照用にのみ提供されています。 Alibaba Cloud アカウントの AccessKey ペアは機密にしてください。

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

   パラメータの説明:

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

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

   • ${endpoint}: OSS のエンドポイント。 エンドポイントは、EMR クラスタが存在するリージョンによって決まります。 EMR クラスタが存在するリージョンで OSS をアクティブ化する必要があります。 詳細については、「OSO リージョンとエンドポイント」をご参照ください。

   このトピックでは、Java コードを使用して Hadoop 公式ウェブサイトの WordCount の例を変更します。 AccessKey ID と AccessKey シークレットの構成がコードに追加されます。 これにより、ジョブに 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}"); // AccessKey ID を設定します
           conf.set("fs.oss.accessKeySecret", "${accessKeySecret}"); // AccessKey シークレットを設定します
           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. コードを JAR ファイルにパッケージ化します。

   上記のコードを記述して保存した後、コードを JAR パッケージに圧縮します。 この例では、onaliyun_mr_wordcount-1.0-SNAPSHOT.jar という名前のパッケージが生成されます。