DataWorks：Buat node EMR MR

Siapkan data awal

Buat file bernama input01.txt yang berisi data awal berikut:

hadoop emr hadoop dw
hive hadoop
dw emr

Unggah file yang menyimpan data awal

1. Masuk ke Konsol OSS. Di panel navigasi kiri, klik Buckets.

2. Di halaman Bucket, temukan bucket yang diinginkan dan klik nama bucket untuk masuk ke halaman Objects.

   Dalam contoh ini, bucket onaliyun-bucket-2 digunakan.

3. Di halaman Objek, klik Create Directory untuk membuat direktori yang digunakan untuk menyimpan data awal dan sumber daya JAR.

   • Atur Directory Name menjadi emr/datas/wordcount02/inputs untuk membuat direktori yang digunakan untuk menyimpan data awal.

   • Atur Directory Name menjadi emr/jars untuk membuat direktori yang digunakan untuk menyimpan sumber daya JAR.

4. Unggah file yang menyimpan data awal ke direktori emr/datas/wordcount02/inputs.

   • Pergi ke direktori /emr/datas/wordcount02/inputs dan klik Upload Object.

   • Di bagian Files to Upload, klik Select Files dan unggah file input01.txt ke bucket.

Gunakan node EMR MR untuk membaca objek OSS dan menghasilkan paket JAR

1. Buka proyek IntelliJ IDEA yang ada dan tambahkan dependensi Project Object Model (POM).

           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-mapreduce-client-common</artifactId>
               <version>2.8.5</version> <!--Versi yang digunakan oleh EMR MR adalah 2.8.5.-->
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>2.8.5</version>
           </dependency>

2. Konfigurasikan parameter berikut untuk membaca data dari dan menulis data ke objek OSS.

   Penting

   Akun Alibaba Cloud memiliki izin untuk memanggil semua operasi API. Jika pasangan AccessKey akun Alibaba Cloud Anda bocor, semua sumber daya di akun Anda mungkin terpapar risiko keamanan tinggi. Kami merekomendasikan agar Anda tidak menyimpan ID AccessKey dan Rahasia AccessKey akun Alibaba Cloud Anda ke dalam kode proyek atau posisi yang mudah ditemukan. Sebagai gantinya, gunakan pengguna RAM untuk memanggil operasi API atau melakukan pemeliharaan rutin O&M. Contoh kode berikut disediakan hanya untuk referensi. Jaga kerahasiaan pasangan AccessKey akun Alibaba Cloud Anda.

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

   Deskripsi Parameter:

   • ${accessKeyId}: ID AccessKey akun Alibaba Cloud Anda.

   • ${accessKeySecret}: Rahasia AccessKey akun Alibaba Cloud Anda.

   • ${endpoint}: Titik Akhir OSS. Titik akhir ditentukan oleh wilayah tempat kluster EMR Anda berada. Anda harus mengaktifkan OSS di wilayah tersebut. Untuk informasi lebih lanjut, lihat Wilayah OSO dan Titik Akhir.

   Dalam topik ini, kode Java digunakan untuk memodifikasi contoh WordCount di situs resmi Hadoop. Konfigurasi ID AccessKey dan Rahasia AccessKey ditambahkan ke kode untuk memberikan izin pekerjaan mengakses objek OSS.

   Contoh Kode

   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. Kemas kode ke dalam file JAR.

   Setelah Anda menulis dan menyimpan kode sebelumnya, kompres kode ke dalam paket JAR. Dalam contoh ini, paket bernama onaliyun_mr_wordcount-1.0-SNAPSHOT.jar dihasilkan.