Pemasangan & Konfigurasi Hadoop (Distribusi Pseudo) dan Latihan: WordCount2

# Clone the repository.
git clone https://github.com/chaos/pdsh.git

# Get into the directory.
cd pdsh

# Run the following commands to compile and install. (Good Luck!)
./bootstrap
./configure
sudo make
sudo make install

### Confirm the installation by verifying its version.
pdsh -V

192.168.0.123   single-node

sudo useradd -m -U -G wheel -s /usr/bin/zsh hadoop

# Set passwd for the user.
sudo passwd hadoop

# Switch to user 'hadoop'.
su - hadoop

tar -xvzf hadoop-3.3.6.tar.gz

export JAVA_HOME="/usr/lib/jvm/java-8-openjdk"
export HADOOP_HOME="$HOME/hadoop-3.3.6"
export HADOOP_CLASSPATH="$JAVA_HOME/lib/tools.jar"
export PATH="$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin"

source ~/.zshrc
cd $HADOOP_HOME/etc/hadoop

export JAVA_HOME="/usr/lib/jvm/java-8-openjdk"

<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>

<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

<configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
    <property>
        <name>mapreduce.application.classpath</name>
        <value>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*:$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*</value>
    </property>
</configuration>

<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
    <property>
        <name>yarn.nodemanager.env-whitelist</name>
        <value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_HOME,PATH,LANG,TZ,HADOOP_MAPRED_HOME</value>
    </property>
</configuration>

hdfs namenode -format

start-dfs.sh && start-yarn.sh

sudo jps

hdfs dfs -mkdir -p /user/name  # Create the user directory. 
hdfs dfs -ls /user/name  # The dir would still be empty. 
hdfs dfs -mkdir -p /user/name/p1_intro/input  # Directories for the test.  

# (local fs environment)
cd $HADOOP_HOME

# Copy files from the local fs to the distributed fs.
hdfs dfs -put etc/hadoop/*.xml /user/name/p1_intro/input

# Run a MapReduce job.
hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.6.jar \
 grep /user/name/p1_intro/input /user/name/p1_intro/output \
 'dfs[a-z.]+'

hdfs dfs -ls -C /user/name/p1_intro
hdfs dfs -ls -C /user/name/p1_intro/output

hdfs dfs -cat /user/name/p1_intro/output/part-r-00000

# OR copy the outputs from the distributed fs to the local fs for observation.
hdfs dfs -get /user/name/p1_intro/output output1

stop-yarn.sh && stop-dfs.sh
exit

# Next login.
ssh localhost
start-dfs.sh && start-yarn.sh

cd ~  # Return to $HOME dir if didn't exit from SSH previously.

# Create the directories for Practice 2.
hdfs dfs -mkdir -p /user/name/p2_wordcount2/input

# Create two files; file1 & file2.
touch file1 file2  # residing in the local fs.
echo 'Hello World, Bye World!' > file1
echo 'Hello Hadoop, Goodbye to hadoop.' > file2

# Move the files from the local fs to the distributed fs.
hdfs dfs -moveFromLocal file1 /user/name/p2_wordcount2/input
hdfs dfs -moveFromLocal file2 /user/name/p2_wordcount2/input

### Confirm the moved files.
hdfs dfs -ls -C /user/name/p2_wordcount2/input

touch patterns.txt

# Create patterns.txt with specified lines.
echo -e '\\.\n\\,\n\\!\nto' > patterns.txt

### Confirm the contents of patterns.txt.
cat patterns.txt

hdfs dfs -moveFromLocal patterns.txt /user/name/p2_wordcount2

### Display directories and files in p2_wordcount2.
hdfs dfs -ls -C /user/name/p2_wordcount2

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", false)) {
        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("Caught exception while parsing the cached file '"
            + 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();
    GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    String[] remainingArgs = optionParser.getRemainingArgs();
    if ((remainingArgs.length != 2) && (remainingArgs.length != 4)) {
      System.err.println("Usage: 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(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
      } else {
        otherArgs.add(remainingArgs[i]);
      }
    }
    FileInputFormat.addInputPath(job, new Path(otherArgs.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1)));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

less WordCount2.java

# Compile the WordCount2 Java source code.
hadoop com.sun.tools.javac.Main WordCount2.java

# Package the compiled Java classes into a JAR file.
jar cf wc.jar WordCount*.class

### List compiled files and jars.
ls ~/ | grep -E 'wc\.jar|WordCount2.*\.class'

hadoop jar wc.jar WordCount2 \
 /user/name/p2_wordcount2/input /user/name/p2_wordcount2/output1

### Check generated files and view output.
hdfs dfs -cat /user/name/p2_wordcount2/output1/part-r-00000

hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true \
 /user/name/p2_wordcount2/input /user/name/p2_wordcount2/output2 \
 -skip /user/name/p2_wordcount2/patterns.txt

### Observe the output.
hdfs dfs -cat /user/name/p2_wordcount2/output2/part-r-00000

hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=false \
 /user/name/p2_wordcount2/input /user/name/p2_wordcount2/output3 \
 -skip /user/name/p2_wordcount2/patterns.txt

### Observe the output.
hdfs dfs -cat /user/name/p2_wordcount2/output3/part-r-00000

stop-yarn.sh && stop-dfs.sh
exit