Panduan Running WordCount.java pada Hadoop

SalesCountry.md

Sales Country

Program menghitung jumlah sales berdasarkan Negara

Sumber Contoh Program : https://www.guru99.com/create-your-first-hadoop-program.html Telah dimodifikasi agar sesuai dengan Hadoop 3.2.0

Sumber Data : SalesJan2009.csv

Tahapan

1. Buat folder pada HDFS untuk menyimpan input dan output files. Jalankan command berikut pada Terminal Ubuntu Server

    hdfs dfs -mkdir -p /sales_country/input
   

2. Buat folder baru pada ubuntu server dengan path ~/workspace/sales

     mkdir ~/workspace
     mkdir ~/workspace/sales
   

3. Download file SalesJan2009.csv ke folder Downloads pada Windows Anda

4. Buka terminal dan masuk ke folder download cd Downloads

5. Updload file SalesJan2009.csv ke Ubuntu VM menggunakan perintah SCP ke dalam folder ~/workspace/sales.

    scp SalesJan2009.csv hadoop@192.168.85.3:~/workspace/sales
   

   Anda dapat memastikan bahwa file telah berhasil terupload ke ubuntu server dengan menjalankan command berikut

    ls -al ~/workspace/sales
   

6. Selanjutnya dapat mengikuti Panduan Menjalankan MapReduce dengan menyesuaikan nama class, jar, dan path folder

Source Code SalesCountry.java

  /*
    Import library dari Java Package 
  */ 
  
  import java.io.IOException; 
  import java.util.StringTokenizer; 
  /*
    Import library dari Hadoop Package untuk menjalankan fungsi Pembacaan, Penulisan File ke dalam HDFS dan
  menjalankan MapReduce 
  */ 
  import org.apache.hadoop.conf.Configuration; 
  import org.apache.hadoop.fs.Path; 
  import org.apache.hadoop.io.IntWritable; 
  import org.apache.hadoop.io.LongWritable; 
  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; 
  
  public class SalesCountry {
  
    public static class SalesMapper extends Mapper <LongWritable, Text, Text, IntWritable> {
    
      private final static IntWritable one = new IntWritable(1);
      
      public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // Membaca 1 baris string dengan tanda , sebagai pemisah
        String valueString = value.toString();
        
        // Memisahkan string menjadi array of string dengan tanda koma sebagai pemisah
        String[] SingleCountryData = valueString.split(",");
        
        // Mengambil data negara, data negara terdapat pada kolom ke 8 atau index 7
        context.write(new Text(SingleCountryData[7]), one);
      }
    }

    public static class SalesCountryReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
      public void reduce(Text t_key, Iterable<IntWritable> values, Context context) throws IOException,
  InterruptedException {
  
        Text key = t_key;
        
        int frequencyForCountry = 0;
        
        for (IntWritable val: values) {
          // frequencyForCountry = frequencyForCountry + val.get()
          frequencyForCountry += val.get();
        }
        context.write(key, new IntWritable(frequencyForCountry));
      }
    }

    // Pada dasarnya penjelasannya sama dengan program WordCount.java, yang menjadi pembeda adalah nama-nama class Mapper, Combiner dan Reducer yang digunakan
    public static void main(String[] args) throws Exception {
          Configuration conf = new Configuration();
          Job job = Job.getInstance(conf, "sales country");
          job.setJarByClass(SalesCountry.class);
          job.setMapperClass(SalesMapper.class);
          job.setCombinerClass(SalesCountryReducer.class);
          job.setReducerClass(SalesCountryReducer.class);
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(IntWritable.class);
          FileInputFormat.addInputPath(job, new Path(args[0]));
          FileOutputFormat.setOutputPath(job, new Path(args[1]));
          System.exit(job.waitForCompletion(true) ? 0 : 1);
      }
  }

Berikut hasil running code tersebut

word_count_mapred.md

Word Count Program

Program menghitung jumlah kata unik memanfaatkan MapReduce pada Hadoop.

Requirements

• Hadoop HDFS dan Yarn berjalan dengan baik
• Semua konfigurasi pada mapred-site.xml telah mengikuti panduan Instalasi Hadoop.
• Sudah mengakses Ubuntu Server melalui terminal Windows menggunakan command ssh -p 22 username@ipubuntuserver Contohnya ssh -p 22 hadoop@192.168.85.3, IP tersebut di dapatkan setelah mengikuti Instalasi Hadoop.

Tahapan

1. Membuat folder input pada HDFS

     hdfs dfs -mkdir -p /word_count/input
   

   

2. Membuat folder untuk menyimpan source code WordCount.java pada Ubuntu Server

     mkdir ~/workspace
     cd ~/workspace
     mkdir word_count
     cd word_count
   

   Full path untuk source code adalah ~/workspace/word_count

3. Membuat file input dengan nama words.txt di dalam folder word_count

     nano words.txt
   

   Kemudian isikan dengan random word atau menggunakan nama Negara. Kemudian Save file tersebut.

   

4. Mengupload file words.txt ke HDFS

     hdfs dfs -put ~/workspace/word_count/words.txt /word_count/input/words.txt
   

   

5. Menuliskan code WordCount.java ke dalam Ubuntu Server menggunakan nano command

     nano ~/workspace/word_count/WordCount.java
   

   Paste code dari situs Hadoop Documentation

   

6. Compile WordCount.java

     cd ~/workspace/word_count/
     hadoop com.sun.tools.javac.Main WordCount.java
   

   Perintah tersebut akan menghasilkan 3 buah file dengan extensi .class

   

7. Generate wc.jar

     jar cf wc.jar WordCount*.class
   

   

8. Run file wc.jar menggunakan Hadoop. Asumsi Input path /word_count/input Output path /word_count/output -> Pastikan folder ini belum ada di HDFS

      hadoop jar wc.jar WordCount /word_count/input /word_count/output
   

   Berikut hasil perintah di atas

     hadoop@dts:~/workspace/word_count$ hadoop jar wc.jar WordCount /word_count/input /word_count/outpput
     2019-08-07 10:45:50,982 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
     2019-08-07 10:45:52,366 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
     2019-08-07 10:45:52,997 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
     2019-08-07 10:45:53,019 INFO mapreduce.JobResourceUploader: Disabling Erasure Coding for path: /tmp/hadoop-yarn/staging/hdfs/.staging/job_1565173480659_0001
     2019-08-07 10:45:53,462 INFO input.FileInputFormat: Total input files to process : 1
     2019-08-07 10:45:53,628 INFO mapreduce.JobSubmitter: number of splits:1
     2019-08-07 10:45:53,815 INFO Configuration.deprecation: yarn.resourcemanager.system-metrics-publisher.enabled is deprecated. Instead, use yarn.system-metrics-publisher.enabled
     2019-08-07 10:45:54,246 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1565173480659_0001
     2019-08-07 10:45:54,250 INFO mapreduce.JobSubmitter: Executing with tokens: []
     2019-08-07 10:45:54,626 INFO conf.Configuration: resource-types.xml not found
     2019-08-07 10:45:54,631 INFO resource.ResourceUtils: Unable to find 'resource-types.xml'.
     2019-08-07 10:45:55,189 INFO impl.YarnClientImpl: Submitted application application_1565173480659_0001
     2019-08-07 10:45:55,347 INFO mapreduce.Job: The url to track the job: http://dts:8088/proxy/application_1565173480659_0001/
     2019-08-07 10:45:55,352 INFO mapreduce.Job: Running job: job_1565173480659_0001
     2019-08-07 10:46:10,808 INFO mapreduce.Job: Job job_1565173480659_0001 running in uber mode : false
     2019-08-07 10:46:10,815 INFO mapreduce.Job:  map 0% reduce 0%
   
     2019-08-07 10:46:20,979 INFO mapreduce.Job:  map 100% reduce 0%
     2019-08-07 10:46:30,107 INFO mapreduce.Job:  map 100% reduce 100%
     2019-08-07 10:46:30,162 INFO mapreduce.Job: Job job_1565173480659_0001 completed successfully
     2019-08-07 10:46:30,367 INFO mapreduce.Job: Counters: 54
             File System Counters
                     FILE: Number of bytes read=106
                     FILE: Number of bytes written=444679
                     FILE: Number of read operations=0
                     FILE: Number of large read operations=0
                     FILE: Number of write operations=0
                     HDFS: Number of bytes read=210
                     HDFS: Number of bytes written=72
                     HDFS: Number of read operations=8
                     HDFS: Number of large read operations=0
                     HDFS: Number of write operations=2
                     HDFS: Number of bytes read erasure-coded=0
             Job Counters
                     Launched map tasks=1
                     Launched reduce tasks=1
                     Data-local map tasks=1
                     Total time spent by all maps in occupied slots (ms)=7917
                     Total time spent by all reduces in occupied slots (ms)=5661
                     Total time spent by all map tasks (ms)=7917
                     Total time spent by all reduce tasks (ms)=5661
                     Total vcore-milliseconds taken by all map tasks=7917
                     Total vcore-milliseconds taken by all reduce tasks=5661
                     Total megabyte-milliseconds taken by all map tasks=8107008
                     Total megabyte-milliseconds taken by all reduce tasks=5796864
             Map-Reduce Framework
                     Map input records=11
                     Map output records=11
                     Map output bytes=141
                     Map output materialized bytes=106
                     Input split bytes=113
                     Combine input records=11
                     Combine output records=7
                     Reduce input groups=7
                     Reduce shuffle bytes=106
                     Reduce input records=7
                     Reduce output records=7
                     Spilled Records=14
                     Shuffled Maps =1
                     Failed Shuffles=0
                     Merged Map outputs=1
                     GC time elapsed (ms)=189
                     CPU time spent (ms)=2370
                     Physical memory (bytes) snapshot=362192896
                     Virtual memory (bytes) snapshot=5402312704
                     Total committed heap usage (bytes)=180224000
                     Peak Map Physical memory (bytes)=232505344
                     Peak Map Virtual memory (bytes)=2697916416
                     Peak Reduce Physical memory (bytes)=129687552
                     Peak Reduce Virtual memory (bytes)=2704396288
             Shuffle Errors
                     BAD_ID=0
                     CONNECTION=0
                     IO_ERROR=0
                     WRONG_LENGTH=0
                     WRONG_MAP=0
                     WRONG_REDUCE=0
             File Input Format Counters
                     Bytes Read=97
             File Output Format Counters
                     Bytes Written=72
     hadoop@dts:~/workspace/word_count$
   
   

   Hasil eksekusi MapReduce dapat dilihat pada HDFS Browser

   

9. Melihat hasil eksekusi WordCount.java

     hdfs dfs -cat /word_count/output/part-r-00000
   

   

WordCount.java

/*
  Baris 5 - 6
  Import library dari Java Package
*/
import java.io.IOException;
import java.util.StringTokenizer;

/*
  Baris 12 - 20
  Import library dari Hadoop Package untuk menjalankan fungsi Pembacaan, Penulisan File ke dalam HDFS dan menjalankan MapReduce
*/
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;

// Deklarasi Class WordCount, nama file sebaiknya sama dengan nama Class.
public class WordCount {

  /*
    Inner class untuk pemetaan kata yang meng-extend class Mapper.
    https://hadoop.apache.org/docs/r3.2.0/api/org/apache/hadoop/mapreduce/Mapper.html
    
    Class Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
    
    Dalam hal ini, Key Input berupa Object, Value Input berupa Text, Key Output Berupa Text dan Value Out berupa Bilangan Bulat.
    
    Berdasarkan percobaan saya mengganti Text dengan String, program MapReduce tidak berhasil dijalankan.
    Kemungkinan besar native type seperti String dan Integer dll tidak dapat langsung digunakan.
    Tipe data yang dapat digunakan dapat dilihat pada link berikut : https://hadoop.apache.org/docs/r3.2.0/api/org/apache/hadoop/io/
    Sepertinya tipe data di Java sudah di ubah pada Hadoop dan diberi akhiran Writeable
  */
  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    // Deklarasi dan instansiasi variable dengan nama one sebagai nilai default untuk jumlah kata yang masuk ke dalam Mapper
    private final static IntWritable one = new IntWritable(1);
    
    // Deklarasi dan instansiasi variable dengan nama word dengan tipe Text yang akan digunakan untuk menyimpan kata pada Mapper
    // Jika ingin mengubah Text menjadi String, maka VALUEIN dan KEYOUT harus menggunakan String agar tipe data nya konsisten.
    private Text word = new Text();

    /*
      Override fungsi map yang ada pada Class Mapper
      
      protected void map(KEYIN key,
                   VALUEIN value,
                   org.apache.hadoop.mapreduce.Mapper.Context context)
                   
      KEYIN, dan VALUEIN harus memiliki tipe yang sama sesuai dengan deklarasi di atas yakni Object dan Text.
     
    */
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      
      /*
        https://docs.oracle.com/javase/7/docs/api/java/util/StringTokenizer.html
        Class StringTokenizer digunakan untuk melakukan pemisahan kata dan menggunakan tanda spasi sebagai pemisah.
        
        Deklarasi dan instansiasi StringTokenizer. itr akan bertipe Iteratable yang dapat ditelusuri menggunakan looping
      */
      StringTokenizer itr = new StringTokenizer(value.toString());
      
      
      // Melakukan looping selama itr masih memiliki token
      while (itr.hasMoreTokens()) {
        // set Text word berdasarkan nilai yang ada pada itr
        word.set(itr.nextToken());
        /*
          Kirimkan data yang bertipe sesuai dengan deklarasi kelas ke konteks MapReduce
          KEYOUT = Text
          VALUEOUT = IntWrteable
        */
        context.write(word, one);
      }
    }
  }

  /*
    Inner class untuk melakukan penghitungan jumlah kata unik yang meng-extend class Reducer.
    https://hadoop.apache.org/docs/r3.2.0/api/org/apache/hadoop/mapreduce/Reducer.html
    
    Class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
    
    KEYIN = Text
    VALUEIN = IntWriteable
    Kedua tipe data tersebut harus sesuai dengan tipe data KEYOUT dan VALUEOUT dari class Mapper
    
    KEYOUT = Text
    VALUEOUT = IntWriteable
  */
  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    
    // Deklarasi dan instansiasi variable dengan nama result untuk menyimpan hasil penjumlahan kata unik
    private IntWritable result = new IntWritable();

    /*
      Override fungsi reduce yang ada pada Class Reducer
      
      protected void reduce(KEYIN key,
                      Iterable<VALUEIN> values,
                      org.apache.hadoop.mapreduce.Reducer.Context context)
                      
      VALUEIN merupakan kumpulan angka yang telah dihasilkan dari proses mapper, dimana satu buah key dapat memiliki beberapa value
    */
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      // Untuk setiap proses reduce, jumlah awal di set menjadi 0
      int sum = 0;
      // loop sebanyak data yang ada pada values
      for (IntWritable val : values) {
        /*
          Lakukan penjumlahan,
          
          sum = sum + val.get()
        */
        sum += val.get();
      }
      
      // Set hasil akhir penjumlahan ke dalam variable result
      result.set(sum);
      
      /*
          Kirimkan data yang bertipe sesuai dengan deklarasi kelas ke konteks MapReduce
          KEYOUT = Text
          VALUEOUT = IntWrteable
        */
      context.write(key, result);
    }
  }

  // Fungsi utama program java
  public static void main(String[] args) throws Exception {
    /*
      http://hadoop.apache.org/docs/r3.2.0/api/org/apache/hadoop/conf/Configuration.html
      Deklarasi dan instansiasi variable conf
      
      Melalui variable conf, kita dapat mengubah configurasi hadoop yang ada pada core-site.xml melalui program.
    */
    Configuration conf = new Configuration();
    /*
      https://hadoop.apache.org/docs/r3.2.0/api/org/apache/hadoop/mapreduce/Job.html
      
      Membuat instance job dengan nama "word count" menggunakan pengaturan default
      
      Melalui variable job tersebut, dapat dilakukan pengaturan, mapper, reducer, tipe key dan output
    */
    Job job = Job.getInstance(conf, "word count");
    // Menetapkan Class utama yang akan dijalankan
    job.setJarByClass(WordCount.class);
    // Menetapkan Mapper class yang akan digunakan untuk proses mapping
    job.setMapperClass(TokenizerMapper.class);
    /*
      Combiner class adalah kelas opsional yang berada di antara Map dan Reduce
      Menetapkan IntSumReducer sebagai combiner class dan reducer class
    */
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    // Tipe data KeyOut dan ValueOut harus di tetapkan pada level job dan sesuai dengan tipe pada output reducer 
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    // Menambahkan path input berdasarkan parameter pertama pada command line saat menjalankan WordCount.java
    FileInputFormat.addInputPath(job, new Path(args[0]));
    // Menambahkan path output berdasarkan parameter kedua pada command line saat menjalankan WordCount.java
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    // Program akan selesai jika job telah selesai
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

error
WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
cat: `/word_count/output/part-r-00000': No such file or directory