We are the world, we are the children
We are the ones who make a brighter day

# 入力ファイルがテキストの場合は単純に Value しか存在しない
入力: (None, "we are the world, we are the children")

出力:

"we", 1
"are", 1
"the", 1
"world", 1
"we", 1
"are", 1
"the", 1
"children", 1

入力: ("we", [1, 1])

出力:

"we", 2

Input: ("we", [2, 1])  # 1行目に "we" は 2回、2行目に "we" は1回出現する

Output:

"we", 3  # 文書内に "we" は3回出現することがわかった！

.
├── Dockerfile
├── main
    ├── WordCount.java  # Hadoop ジョブ（Java）
    ├── scripts  # Hadoop 起動スクリプトなど
    │   ├── create-input-text.sh
    │   ├── execute-wordcount-python.sh
    │   ├── execute-wordcount.sh
    │   ├── make-jar.sh
    │   └── start-hadoop.sh
    └── streaming  # Hadoop Streaming ジョブ（Python）
        └── python
            ├── map.py
            └── reduce.py

FROM centos:centos7

RUN yum -y update
RUN yum -y install sudo

# インストール：JDK
RUN yum -y install java-1.8.0-openjdk java-1.8.0-openjdk-devel

# 環境変数の設定
ENV JAVA_HOME /usr/lib/jvm/java-1.8.0-openjdk
ENV PATH $PATH:$JAVA_HOME/bin
# hadoop コマンドを実行できるようにするため（また、tools.jar は javac コンパイラを含む）
ENV HADOOP_CLASSPATH $JAVA_HOME/lib/tools.jar

# インストール：CDH 5 パッケージ
RUN rpm --import http://archive.cloudera.com/cdh5/redhat/6/x86_64/cdh/RPM-GPG-KEY-cloudera # GPG キーのインストール
RUN rpm -ivh http://archive.cloudera.com/cdh5/one-click-install/redhat/6/x86_64/cloudera-cdh-5-0.x86_64.rpm # yum リポジトリへ登録
RUN yum -y install hadoop-conf-pseudo # 疑似分散モードの設定と YARN や HDFS などを提供するパッケージをインストール

ADD main main
RUN chmod +x -R main

WORKDIR main

# コマンド実行後もコンテナを起動させ続ける
CMD ["tail", "-f", "/dev/null"]

docker image build -t {名前空間/イメージ名:タグ名} .

docker container run --name {コンテナ名} -d -p 50070:50070 {名前空間/イメージ名:タグ名}

docker exec -it {コンテナ名} /bin/bash

#!/usr/bin/env bash

# NameNode が管理するメタデータ領域をフォーマット
# 初めて HDFS を利用する際には必要なコマンド
sudo -u hdfs hdfs namenode -format

# HDFS の起動
for x in `cd /etc/init.d ; ls hadoop-hdfs-*` ; do sudo service $x start ; done

# 必要なディレクトリを作成
sudo /usr/lib/hadoop/libexec/init-hdfs.sh

# HDFS のファイル構造を把握
# hadoop fs は HDFS を操作するコマンド（ここでは HDFS 上で ls -R / を実行しています）
sudo -u hdfs hadoop fs -ls -R /

# YARN の起動
sudo service hadoop-yarn-resourcemanager start
sudo service hadoop-yarn-nodemanager start
sudo service hadoop-mapreduce-historyserver start

[root@xxxxxxxxx main]# ./scripts/start-hadoop.sh

<configuration>
  ...
  <!-- HDFS (NameNode) の接続先の設定 -->
  <property>
    <name>fs.defaultFS</name>
    <!-- 疑似分散モードの場合：hdfs スキームを用いることで、ローカルファイルシステムではなく HDFS を利用するように設定できる -->
    <value>hdfs://localhost:8020</value>
    <!-- 完全分散モードの場合：NameNode と DataNode のサーバーが違うため、名前解決を行う必要がある -->
    <!-- <value>hdfs://nn.example.com:8020</value> -->
  </property>
  ...
</configuration>

<!-- HDFS の設定 -->
<configuration>
  ...
  <!-- レプリカ数の設定 -->
  <property>
    <name>dfs.replication</name>
    <!-- 疑似分散モードの場合：サーバーが 1 台なのでこれ以上にはできない -->
    <value>1</value>
    <!-- 完全分散モードの場合：デフォルトは 3 -->
    <!-- <value>3</value> -->
  </property>
  ...
</configuration>

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordCount extends Configured implements Tool {
    /**
     * Mapper<入力キーの型, 入力バリューの型, 出力キーの型, 出力バリューの型> を継承したクラス.
     */
    public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        public void setup(Context context) throws IOException, InterruptedException {
            // 初期化処理
        }

        /**
         * Map 処理を記述する.
         *
         * @param key その行が先頭から何バイト目の位置にあるかというバイトオフセット値（通常は利用しない）
         * @param value 1行分のデータ
         * @param context Context を通してジョブの設定や入出力データにアクセス可能
         */
        @Override
        public void map(LongWritable 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);
            }
        }

        @Override
        public void cleanup(Context context) throws IOException, InterruptedException {
            // クリーンナップ処理
        }
    }

    /**
     * Reducer<入力キーの型, 入力バリューの型, 出力キーの型, 出力バリューの型> を継承したクラス.
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        @Override
        public void setup(Context context) throws IOException, InterruptedException {
            // 初期化処理
        }

        /**
         * Reduce 処理を記述する.
         *
         * @param key Map 処理の出力（キー）
         * @param values Map 処理の出力（バリューのイテラブル）
         * @param context Context
         */
        @Override
        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);
        }

        @Override
        public void cleanup(Context context) throws IOException, InterruptedException {
            // クリーンナップ処理
        }
    }

    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(2);
        }

        // JobTracker に対してジョブを投入する
        Job job = Job.getInstance(getConf(), "WordCount");

        // 入力データに応じて自動的に数が決まるMapタスクとは異なり、Reduceタスクの数は自分で指定する必要がある
        job.setNumReduceTasks(2);

        // jarファイルに格納されたクラスのうちの１つを指定する
        job.setJarByClass(WordCount.class);

        // Mapper、Combiner、Reducer としてどのクラスを利用するか指定する
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // テキストファイルからデータの入出力を行う
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        // 入出力用のディレクトリのパス
        TextInputFormat.addInputPath(job, new Path(args[0]));
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        // ジョブの完了を待つ
        return (job.waitForCompletion(true) ? 0 : 1);
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new WordCount(), args);
        System.exit(res);
    }
}

#!/usr/bin/env bash

# コンパイル
hadoop com.sun.tools.javac.Main WordCount.java

# jar の作成
jar cf wc.jar WordCount*.class

[root@xxxxxxxxx main]# ./scripts/make-jar.sh

#!/usr/bin/env bash

# 入力のテキストファイルを作成
echo "apple lemon apple lemon lemon grape" > input.txt

# 入力のテキストファイルを HDFS に配置
sudo -u hdfs hadoop fs -mkdir -p /user/hdfs/input
sudo -u hdfs hadoop fs -put input.txt /user/hdfs/input

[root@xxxxxxxxx main]# ./scripts/create-input-text.sh

#!/usr/bin/env bash

# WordCount.java の実行
# hadoop jar {jar ファイルのパス} {メインクラス名} {入力ファイルのパス} {出力先のパス}
sudo -u hdfs hadoop jar wc.jar WordCount /user/hdfs/input/input.txt /user/hdfs/output01

# 結果の表示
sudo -u hdfs hadoop fs -ls /user/hdfs/output01
sudo -u hdfs hadoop fs -cat /user/hdfs/output01/part-r-*

[root@xxxxxxxxx main]# ./scripts/execute-wordcount.sh

apple   2
grape   1
lemon   3

#!/usr/bin/env bash

# Hadoop Streaming の実行
sudo -u hdfs hadoop jar /usr/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming-2.6.0-mr1-cdh5.16.2.jar \
-input /user/hdfs/input/input.txt -output /user/hdfs/output02 \
-mapper /main/streaming/python/map.py -reducer /main/streaming/python/reduce.py

# 結果の表示
sudo -u hdfs hadoop fs -ls /user/hdfs/output02
sudo -u hdfs hadoop fs -cat /user/hdfs/output02/part-*

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import re
import sys


# 1行を空文字区切りで分割し、(単語, 1) のキーバリューを生成する
def map_fn(line):
    return [(key, 1) for key in re.split(r'\s', line.strip()) if key]

# キーバリューを標準出力に出力する
def output(records):
    for key, value in records:
        print '{0}\t{1}'.format(key, value)

# 標準入力から入力を受け取る
for l in sys.stdin:
    output(map_fn(l))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import re
import sys

results = {}


# 単語の出現回数を数える
def reduce_fn(line):
    key, value = re.split(r'\t', line.strip())
    if not key in results:
        results[key] = 0
    results[key] = results[key] + int(value)

# キーバリュー（最終的な処理結果）を標準出力に出力する
def output(records):
    for k, v in records:
        print '{0}\t{1}'.format(k, v)

# 標準入力から map 処理の出力を受け取る
for l in sys.stdin:
    reduce_fn(l)
output(sorted(results.items()))

apple   2
grape   1
lemon   3