Hadoop 笔记

useradd -m hadoop -s /bin/bash
passwd hadoop
visudo #添加到sudo组
#配置ssh登录限制（安全）、将hadoop限制为仅限localhost登录
#配置免密可能无法登录
# vi /etc/ssh/sshd_config
# service sshd restart

#配置免密登录（通过秘钥）
cd ~/.ssh/                     # 若没有该目录，请先执行一次ssh localhost
ssh-keygen -t rsa              # 会有提示，都按回车就可以
cat ./id_rsa.pub >> ./authorized_keys  # 加入授权
chmod 600 authorized_keys id_rsa

# 修改SSH客户端配置
sudo vim /etc/ssh/ssh_config
# 修改连接配置
# GSSAPIAuthentication yes
# StrictHostKeyChecking no

# 修改SSH服务器配置
sudo vim /etc/ssh/sshd_config
# UseDNS no
sudo systemctl restart sshd

# 安装Java

cd ~
mkdir java
cd java
# 下载
wget --no-cookies --header "Cookie: oraclelicense=accept-securebackup-cookie;" https://download.oracle.com/otn-pub/java/jdk/8u201-b09/42970487e3af4f5aa5bca3f542482c60/jdk-8u201-linux-x64.tar.gz
# 解压
tar -xzvf jdk-8u201-linux-x64.tar.gz
# 创建软链接
ln -s jdk1.8.0_201 default
# 编辑环境变量
vim ~/.bashrc
## Java
#export JAVA_HOME=/home/bigdata/java/default
#export CLASSPATH=".:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar:$CLASSPATH"
#export PATH="$JAVA_HOME/bin:$PATH"

# 立即生效环境变量
source ~/.bashrc
# 查看安装是否成功
java -version

# 安装Hadoop
cd ~
mkdir hadoop
cd hadoop
wget http://mirror.bit.edu.cn/apache/hadoop/common/hadoop-2.6.5/hadoop-2.6.5.tar.gz
tar -xzvf hadoop-2.6.5.tar.gz
ln -s hadoop-2.6.5 default
vim ~/.bashrc
# 添加如下环境变量
## Hadoop
#export HADOOP_HOME=/home/bigdata/hadoop/default
#export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
#export PATH="$HADOOP_HOME/bin:$PATH"
#export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib:$HADOOP_COMMON_LIB_NATIVE_DIR"
source ~/.bashrc
hadoop

mkdir input
cp etc/hadoop/*.xml input
hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.8.0.jar grep input output 'dfs[a-z.]+'
cat output/*
#输出结果为1	dfsadmin表示成功

# cd hadoop目录
cd $HADOOP_HOME

vim etc/hadoop/core-site.xml
#添加如下
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>fs.defaultFS</name>
    <value>hdfs://localhost/</value> <!-- 默认端口为8200 -->
  </property>

  <!-- mac或Window设置（没有手动编译Hadoop本地库情况下）-->
  <property>
    <name>io.native.lib.available</name>
    <value>false</value>
  </property>

  <!-- 使用 HADOOP_CONF_DIR 环境变量指定配置文件目录 -->
</configuration>

vim etc/hadoop/hdfs-site.xml
#添加如下
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
  <!-- 默认的Hadoop存储目录为/tmp/hadoop-${user.name} -->
  <!-- <property>
    <name>dfs.namenode.name.dir</name>
    <value>file:/usr/local/hadoop/tmp/dfs/name</value>
  </property><property>
    <name>dfs.datanode.data.dir</name>
    <value>file:/usr/local/hadoop/tmp/dfs/data</value>
  </property> -->
</configuration>

./bin/hdfs namenode -format

./sbin/start-dfs.sh

jps

./sbin/stop-dfs.sh   # 关闭
rm -r ./tmp     # 删除 tmp 文件，注意这会删除 HDFS 中原有的所有数据
./bin/hdfs namenode -format   # 重新格式化 NameNode
./sbin/start-dfs.sh  # 重启

#上面的单机模式，grep 例子读取的是本地数据，伪分布式读取的则是 HDFS 上的数据。要使用 HDFS，首先需要在 HDFS 中创建用户目录：
./bin/hdfs dfs -mkdir -p /user/hadoop

#接着将 ./etc/hadoop 中的 xml 文件作为输入文件复制到分布式文件系统中，即将 /usr/local/hadoop/etc/hadoop 复制到分布式文件系统中的 /user/hadoop/input 中。我们使用的是 hadoop 用户，并且已创建相应的用户目录 /user/hadoop ，因此在命令中就可以使用相对路径如 input，其对应的绝对路径就是 /user/hadoop/input:
./bin/hdfs dfs -mkdir input
./bin/hdfs dfs -put ./etc/hadoop/*.xml input

#查看文件列表
./bin/hdfs dfs -ls input

#运行
./bin/hadoop jar ./share/hadoop/mapreduce/hadoop-mapreduce-examples-*.jar grep input output 'dfs[a-z.]+'
#查看运行结果
./bin/hdfs dfs -cat output/*
#取回本地
rm -r ./output    # 先删除本地的 output 文件夹（如果存在）
./bin/hdfs dfs -get output ./output     # 将 HDFS 上的 output 文件夹拷贝到本机
cat ./output/*

./bin/hdfs dfs -rm -r output    # 删除 output 文件夹

./sbin/stop-dfs.sh

cp ./etc/hadoop/mapred-site.xml.template ./etc/hadoop/mapred-site.xml
vim etc/hadoop/yarn-site.xml
<?xml version="1.0"?>
<!-- yarn-site.xml -->
<configuration>
  <property>
    <name>yarn.resourcemanager.hostname</name>
    <value>localhost</value>
  </property>
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle</value>
  </property>
</configuration>

vim etc/hadoop/mapred-site.xml
<?xml version="1.0"?>
<!-- mapred-site.xml -->
<configuration>
  <property>
    <name>mapreduce.framework.name</name>
    <value>yarn</value>
  </property>
</configuration>

./sbin/start-dfs.sh
./sbin/start-yarn.sh      # 启动YARN
./sbin/mr-jobhistory-daemon.sh start historyserver  # 开启历史服务器，才能在Web中查看任务运行情况
jps

http://localhost:8088/cluster

192.168.3.20  hadoop-master
192.168.3.21  hadoop-slave1
192.168.3.22  hadoop-slave2

<configuration>
        <property>
                <name>hadoop.tmp.dir</name>
                <value>file:/usr/hadoop/default/tmp</value>
                <description>Abase for other temporary directories.</description>
        </property>
        <property>
                <name>fs.defaultFS</name>
                <value>hdfs://hadoop-master:9000</value>
        </property>
</configuration>

export JAVA_HOME=/usr/java/default
export HADOOP_OPTS="-Djava.library.path=${HADOOP_COMMON_LIB_NATIVE_DIR}"

<configuration>
    <property>
        <name>dfs.replication</name>
        <value>2</value>
    </property>
    <property>
        <name>dfs.namenode.secondary.http-address</name>
        <value>hadoop-slave1:50090</value>
    </property>
    <property>
        <name>dfs.namenode.name.dir</name>
        <value>file:/usr/hadoop/default/tmp/dfs/name</value>
    </property>
    <property>
        <name>dfs.datanode.data.dir</name>
        <value>file:/usr/hadoop/default/tmp/dfs/data</value>
    </property>
</configuration>

hadoop-master
hadoop-slave1
hadoop-slave2

export JAVA_HOME=/usr/java/default

<configuration>
    <property>
     <name>mapreduce.framework.name</name>
     <value>yarn</value>
    </property>
</configuration>

export JAVA_HOME=/usr/java/default

<configuration>
     <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
     </property>
     <property>
        <name>yarn.resourcemanager.hostname</name>
        <value>hadoop-slave1</value>
     </property>
</configuration>

hdfs namenode -format

# 启动
start-dfs.sh
# 终止
stop-dfs.sh

#启动
start-yarn.sh
#终止
stop-yarn.sh

	URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
	in = new URL('file:///etc/hosts').openStream();

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);

// 读写
InputStream in = fs.open(new Path(uri)); // 获取一个输入流
fs.exists() // 判断文件是否存在
OutputStream out = fs.create(new Path(dst), new Progressable() { // 创建一个文件并获取一个输出流（存在报错）
	public void progress() {
		System.out.print(".");
	}
});
OutputStream out1 = fs.append(new Path(dst)) //追加一个文件内容

// 创建目录
fs.mkdirs(new Path(dir)) // 创建一个目录

// 文件信息
fs.getFileStatus(new Path(dst)) //查看文件元数据（文件信息）
fs.listStatus(new Path(dir))  // 查看目录下的文件信息
fs.globStatus(new Path(globPath))	// 通过glob模式获取文件信息

// 删除
fs.delete(new Path(p))

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class FileSystemCat {
	public static void main(String[] args) throws IOException {
		String uri = args[0]; //获取参数
		Configuration conf = new Configuration(); //创建一个hadoop配置对象
		FileSystem fs = FileSystem. get(URI.create (uri), conf); //获取文件系统实例
		InputStream in = null;
		try {
			in = fs.open(new Path(uri)); //打开一个文件流
			IOUtils.copyBytes(in, System.out, 4096, false); //将文件的前4096个字节打印到标准输出
		} finally {
			IOUtils.closeStream(in);
		}
	}
}

start-all.sh
#stop-all.sh

cd /usr/local/hadoop

mkdir myclass
mkdir input

cd input
touch  quangle.txt
vim quangle.txt
#复制如下内容进入
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.

hadoop fs -mkdir /class4
hadoop fs -ls /

hadoop fs -put quangle.txt /class4
hadoop fs -ls /class4

cd ../myclass/
vi FileSystemCat.java
#将代码复制到文件中

#编译
javac -classpath ../share/hadoop/common/hadoop-common-2.8.0.jar FileSystemCat.java
#打包
jar cvf FileSystemCat.jar FileSystemCat.class
ls
#输出jar和class文件

hadoop jar FileSystemCat.jar FileSystemCat /class4/quangle.txt

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

public class LocalFile2Hdfs {
	public static void main(String[] args) throws IOException {
		String local = args[0]; //本地文件路径
        String uri = args[1]; //hdfs系统的路径

        FileInputStream in = null;
        OutputStream out = null;
        Configuration conf = new Configuration();

        try {
        	// 从本地文件读取文件
        	in = new FileInputStream(new File(local));

        	//目标文件配置
        	FileSystem fs = FileSystem.get(URI.create(uri), conf);
        	out = fs.create(new Path(uri), new Progressable() {
                @Override
                public void progress() {
                    System.out.println("*");
                }
            });

        	// 跳过前100个字符
            in.skip(100);

            byte[] buffer = new byte[20];
            // 从101的位置读取20个字符到buffer中
            int bytesRead = in.read(buffer);
            if (bytesRead >= 0) { //数据写入系统
                out.write(buffer, 0, bytesRead);
            }

        } finally {
        	IOUtils.closeStream(in);
            IOUtils.closeStream(out);
		}

	}
}

cd /usr/local/hadoop/myclass
vim LocalFile2Hdfs.java
#将代码复制到文件中
cd ../input
vim local2hdfs.txt
#复制测试文本

#编译打包运行
cd ../myclass
javac -classpath ../share/hadoop/common/hadoop-common-2.8.0.jar LocalFile2Hdfs.java
#打包时注意，由于使用了匿名对象所以生成了LocalFile2Hdfs\$1.class
jar cvf LocalFile2Hdfs.jar LocalFile2Hdfs.class LocalFile2Hdfs\$1.class
hadoop jar LocalFile2Hdfs.jar LocalFile2Hdfs ../input/local2hdfs.txt /class4/local2hdfs_part.txt

#查看
hadoop fs -cat /class4/local2hdfs_part.txt

import java.io.FileOutputStream;
import java.io.OutputStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class Hdfs2LocalFile {
	public static void main(String[] args) throws Exception {

		String uri = args[0];
		String local = args[1];

		FSDataInputStream in = null;
		OutputStream out = null;
		Configuration conf = new Configuration();
		try {
			FileSystem fs = FileSystem.get(URI.create(uri), conf);
			in = fs.open(new Path(uri));
			out = new FileOutputStream(local);

			byte[] buffer = new byte[20];
			in.skip(100);
			int bytesRead = in.read(buffer);
			if (bytesRead >= 0) {
				out.write(buffer, 0, bytesRead);
			}
		} finally {
			IOUtils.closeStream(in);
			IOUtils.closeStream(out);
		}
	}
}

cd /usr/local/hadoop/myclass
vim Hdfs2LocalFile.java
#将代码复制到文件中
cd ../input
vim hdfs2local.txt
#复制测试文本

#上传测试文本
hadoop fs -copyFromLocal hdfs2local.txt /class4/hdfs2local.txt
hadoop fs -ls /class4/

#编译打包运行
cd ../myclass
javac -classpath ../share/hadoop/common/hadoop-common-2.8.0.jar Hdfs2LocalFile.java
#打包
jar cvf Hdfs2LocalFile.jar Hdfs2LocalFile.class
hadoop jar Hdfs2LocalFile.jar Hdfs2LocalFile  /class4/hdfs2local.txt hdfs2local_part.txt

#查看
cat hdfs2local_part.txt

root
  |-- prod
  |-- dev
          |-- eng
          |-- science

<configuration>
  <property>
    <name>yarn.resourcemanager.scheduler.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
  </property>
<configuration>

<?xml version="1.0"?>
<configuration>
  <property>
    <name>yarn.scheduler.capacity.root.queues</name>
    <value>prod,dev</value>
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.dev.queues</name>
    <value>eng,science</value>
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.prod.capacity</name>
    <value>40</value> <!-- prod占40% -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.dev.capacity</name>
    <value>60</value> <!-- dev占60% -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.dev.maximum-capacity</name>
    <value>75</value> <!-- dev的最大容量为75% -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.dev.eng.capacity</name>
    <value>50</value><!-- eng占dev的50% -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.dev.science.capacity</name>
    <value>50</value><!-- science占dev的50% -->
  </property>
</configuration>

map: (K1, V1) -> list(K2, V2)
reduce: (K2, list(V2)) -> list(K3, V3)

# 不使用JAR方式
export HADOOP_CLASSPATH=target/experiment.jar
hadoop cn.rectcircle.hadooplearn.mrunit.MaxTemperatureDriver -conf src/main/resources/conf/hadoop-local.xml input/ncdc/micro output

# 使用jar，命令运行
unset HADOOP_CLASSPATH
hadoop jar target/experiment.jar cn.rectcircle.hadooplearn.mrunit.MaxTemperatureDriver -conf src/main/resources/conf/hadoop-localhost.xml input/ncdc/all max-temp

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>cn.rectcircle.hadooplearn</groupId>
  <artifactId>wordcount</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>wordcount</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <hadoop.version>2.6.5</hadoop.version>
  </properties>


  <dependencies>
    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>${hadoop.version}</version>
      <scope>provided</scope>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>${hadoop.version}</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>${hadoop.version}</version>
    </dependency>

    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.12</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <finalName>${project.artifactId}</finalName>
  </build>

</project>

package cn.rectcircle.hadooplearn.wordcount;

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 WordCount {

	//Map过程继承Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
	//实现map方法
	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		//需要覆写的函数，Mapper的主体
		//声明为void map(KEYIN key, VALUEIN value, Context context)
		//用户输入的数据源使用InputSplit进行分片，存储每个分片的长度和偏移量的数组
		//参数key、value由InputFormat针对每个分片产生，默认的InputFormat为TextInputFormat,行为为按照换行进行分割，结果为<分片偏移量, line>
		//结果写入context即可
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			//默认情况下value为一行数据
			// StringTokenizer创建一个类似迭代器的对象，按照字符串的制表回车换行的字符分割
			StringTokenizer itr = new StringTokenizer(value.toString());
			while (itr.hasMoreTokens()) {
				//设置Text，Text应该是一个可变对象
				word.set(itr.nextToken());
				//将数据写入上下文
				context.write(word, one);
			}
		}
	}

	//Reduce过程继承自Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
	public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		// 实现reduce方法 void reduce(KEYIN key, Iterable<VALUEIN> values, Context context)
		// 将map的结果按照key进行汇总到values里面进行reduce操作
		// key、value为map的输出
		// 结果写入context，然后经过OutFormat输出到文件
		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> <out>");
			System.exit(2);
		}
		//创建一个作业
		Job job = Job.getInstance(conf, "word count");
		//设置工作的类
		job.setJarByClass(WordCount.class);
		//设定Mapper
		job.setMapperClass(TokenizerMapper.class);
		//设定Combiner
		job.setCombinerClass(IntSumReducer.class);
		//设置Reduce
		job.setReducerClass(IntSumReducer.class);
		//设定输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		//设置输入输出路径
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		//等待工作完成退出
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

mvn package

mkdir input
echo 'Hello Hadoop Goodbye Hadoop' > input/file01
echo 'Hello Hadoop Goodbye Hadoop' > input/file02
hadoop fs -mkdir -p exp01/input
hadoop fs -put input/* exp01/input
hadoop fs -ls exp01/input

hadoop jar wordcount.jar cn.rectcircle.hadooplearn.wordcount.WordCount exp01/input exp01/output
hadoop fs -ls exp01/output
hadoop fs -cat exp01/output/part-r-00001

package cn.rectcircle.hadooplearn.wordcount;

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;

		// 启动map之前执行的操作
		@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);
				}
			}
		}

		//解析skip文件
		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, "");
			}
			//Map逻辑
			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 = Job.getInstance(conf, "word count");
		//通过查找给定类的来源来设置Jar。
		job.setJarByClass(WordCount2.class);
		//设置Mapper
		job.setMapperClass(TokenizerMapper.class);
		//设定Combiner
		job.setCombinerClass(IntSumReducer.class);
		//设定Reduce
		job.setReducerClass(IntSumReducer.class);
		//设定输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		//查找skip参数
		List<String> otherArgs = new ArrayList<String>();
		for (int i = 0; i < remainingArgs.length; ++i) {
			if ("-skip".equals(remainingArgs[i])) {
				//如果存在skip参数
				//将该路径加入Job缓存文件
				job.addCacheFile(new Path(remainingArgs[++i]).toUri());
				//并设置Job配置
				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)));
		//启动Job并等待完成并退出
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

mvn package

echo 'Hello World, Bye World!' > file01
echo 'Hello Hadoop, Goodbye to hadoop.' > file02
hadoop fs -mkdir -p exp02/input
hadoop fs -put file0* exp02/input
$ cat patterns.txt
\.
\,
\!
to
hadoop fs -put patterns.txt exp02/input

# 不带参数运行
hadoop jar wordcount.jar cn.rectcircle.hadooplearn.wordcount.WordCount2 exp02/input exp02/output
hadoop fs -cat exp02/output/part-r-00000
# 带参数运行
hadoop jar wordcount.jar cn.rectcircle.hadooplearn.wordcount.WordCount2 exp02/input exp02/output2 -skip exp02/input/patterns.txt
hadoop fs -cat exp02/output2/part-r-00000

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

import re
import sys

for line in sys.stdin:
  val = line.strip()
  (year, temp, q) = (val[15:19], val[87:92], val[92:93])
  if (temp != "+9999" and re.match("[01459]", q)):
    print "%s\t%s" % (year, temp)

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

import sys

(last_key, max_val) = (None, -sys.maxint)
for line in sys.stdin:
  (key, val) = line.strip().split("\t")
  if last_key and last_key != key:
    print "%s\t%s" % (last_key, max_val)
    (last_key, max_val) = (key, int(val))
  else:
    (last_key, max_val) = (key, max(max_val, int(val)))

if last_key:
  print "%s\t%s" % (last_key, max_val)

cat input/ncdc/sample.txt | \
src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_map.py | \
sort | \
src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_reduce.py

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
  -input input/ncdc/sample.txt \
  -output output \
  -mapper src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_map.py \
  -combiner src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_reduce.py \
  -reducer src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_reduce.py
  # 分布式环境下需要将制定files
  # -files src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_map.py, \
  #        src/main/java/cn/rectcircle/hadooplearn/mrintro/stream/python/max_temperature_reduce.py \