tar -zxf jdk-7u45-linux-x64.gz
mv jdk1.7.0_45 /usr/local/jdk1.7

export JAVA_HOME=/usr/local/jdk1.7
export CLASSPATH=$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:.
export PATH=$JAVA_HOME/bin:$PATH

java -version

java version "1.7.0_45"
Java(TM) SE Runtime Environment (build 1.7.0_45-b18)
Java HotSpot(TM) 64-Bit Server VM (build 24.45-b08, mixed mode)

http://www.apache.org/dyn/closer.cgi/hadoop/common/
http://mirror.bit.edu.cn/apache/hadoop/common/stable/hadoop-2.9.1.tar.gz
tar -zxf hadoop-2.9.1.tar.gz
mv hadoop-2.9.1 /usr/local/hadoop2.9.1

export JAVA_HOME=/usr/local/jdk1.7
export HADOOP_HOME=/usr/local/hadoop2.9.1
export CLASSPATH=$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:.
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$JAVA_HOME/bin:$PATH

hadoop version

Hadoop 2.9.1
Subversion https://github.com/apache/hadoop.git -r e30710aea4e6e55e69372929106cf119af06fd0e
Compiled by root on 2018-04-16T09:33Z
Compiled with protoc 2.5.0
From source with checksum 7d6d2b655115c6cc336d662cc2b919bd
This command was run using /usr/local/hadoop2.9.1/share/hadoop/common/hadoop-common-2.9.1.jar

[root@localhost www]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa): 回车
Enter passphrase (empty for no passphrase): 回车
Enter same passphrase again: 回车
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
11:00:39:f3:8b:d7:19:47:1e:67:06:7e:cd:9f:a0:4f root@localhost

[root@localhost sbin]# cd ~/.ssh/
[root@localhost .ssh]# cat id_rsa.pub >> authorized_keys

[root@localhost .ssh]# ssh root@localhost

systemctl disable firewalld.service
systemctl stop firewalld.service

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
	<property>
			<name>fs.defaultFS</name>
			<value>hdfs://localhost:9000</value>
			<description>NameNode</description>
	</property>
	<property>
    		<name>hadoop.tmp.dir</name>
    		<value>/usr/local/hadoop2.9.1/data</value>
  	</property>
</configuration>

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
	<property>
		<name>mapreduce.framework.name</name>
		<value>yarn</value>
	</property>
	<property>  
        <name>mapreduce.jobhistory.address</name>  
        <value>0.0.0.0:10020</value>  
	</property>
</configuration>

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

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

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

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

hdfs namenode -format
cd /usr/local/hadoop2.9.1/sbin
start-dfs.sh
start-yarn.sh

[root@localhost sbin]# jps
6738 ResourceManager
6404 DataNode
7032 Jps
6577 SecondaryNameNode
6267 NameNode
6855 NodeManager

[root@localhost sbin]# netstat -openlut
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       User       Inode      PID/Program name     Timer
tcp        0      0 0.0.0.0:50020           0.0.0.0:*               LISTEN      0          49246      6404/java            off (0.00/0/0)
tcp        0      0 127.0.0.1:9000          0.0.0.0:*               LISTEN      0          47962      6267/java            off (0.00/0/0)
tcp        0      0 0.0.0.0:50090           0.0.0.0:*               LISTEN      0          50280      6577/java            off (0.00/0/0)
tcp        0      0 0.0.0.0:50070           0.0.0.0:*               LISTEN      0          47955      6267/java            off (0.00/0/0)
tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      0          15627      1135/sshd            off (0.00/0/0)
tcp        0      0 127.0.0.1:25            0.0.0.0:*               LISTEN      0          16691      1545/master          off (0.00/0/0)
tcp        0      0 0.0.0.0:50010           0.0.0.0:*               LISTEN      0          48972      6404/java            off (0.00/0/0)
tcp        0      0 0.0.0.0:50075           0.0.0.0:*               LISTEN      0          49243      6404/java            off (0.00/0/0)
tcp        0      0 127.0.0.1:55740         0.0.0.0:*               LISTEN      0          48978      6404/java            off (0.00/0/0)
tcp6       0      0 127.0.0.1:8033          :::*                    LISTEN      0          55050      6738/java            off (0.00/0/0)
tcp6       0      0 :::8040                 :::*                    LISTEN      0          54139      6855/java            off (0.00/0/0)
tcp6       0      0 :::8042                 :::*                    LISTEN      0          55047      6855/java            off (0.00/0/0)
tcp6       0      0 :::22                   :::*                    LISTEN      0          15629      1135/sshd            off (0.00/0/0)
tcp6       0      0 127.0.0.1:8088          :::*                    LISTEN      0          54137      6738/java            off (0.00/0/0)
tcp6       0      0 :::40377                :::*                    LISTEN      0          54131      6855/java            off (0.00/0/0)
tcp6       0      0 ::1:25                  :::*                    LISTEN      0          16692      1545/master          off (0.00/0/0)
tcp6       0      0 :::13562                :::*                    LISTEN      0          55046      6855/java            off (0.00/0/0)
tcp6       0      0 127.0.0.1:8030          :::*                    LISTEN      0          51406      6738/java            off (0.00/0/0)
tcp6       0      0 127.0.0.1:8031          :::*                    LISTEN      0          51384      6738/java            off (0.00/0/0)
tcp6       0      0 127.0.0.1:8032          :::*                    LISTEN      0          51956      6738/java            off (0.00/0/0)
udp        0      0 0.0.0.0:68              0.0.0.0:*                           0          21329      3830/dhclient        off (0.00/0/0)
udp        0      0 0.0.0.0:46650           0.0.0.0:*                           0          21256      3830/dhclient        off (0.00/0/0)
udp6       0      0 :::43191                :::*                                0          21257      3830/dhclient        off (0.00/0/0)

http://192.168.111.111:50070/
http://192.168.111.111:8088/

#!/bin/bash
for i in {1901..2018}
do
cd /data/soft/ncdc
wget --execute robots=off -r -np -nH --cut-dirs=4 -R index.html* ftp://ftp.ncdc.noaa.gov/pub/data/noaa/$i/
done

#!/bin/bash

# 处理全部数据
for file in ncdc/*.gz
do
  echo "$file"
  gunzip -c $file >> /home/www/bigdata/target/all.txt
done

# -put 后的 - 表示从标准输出中读取,也就是接受前面压缩后通过管道传过来的数据
# gzip /home/www/bigdata/target/all.txt | hadoop fs -put - /home/hdfs/
# 也可以分成两步:
gzip /home/www/bigdata/target/all.txt
hadoop fs -put /home/www/bigdata/target/all.txt.gz /home/hdfs/

#!/bin/bash

# 只处理 4 年的数据
for file in `ls ncdc | grep -E '1901.gz|1902.gz|1903.gz|1904.gz'`
do
  echo "$file"
  gunzip -c ncdc/$file >> /home/www/bigdata/target/fouryear.txt
done

# -put 后的 - 表示从标准输出中读取,也就是接受前面压缩后通过管道传过来的数据
# gzip /home/www/bigdata/target/fouryear.txt | hadoop fs -put - /home/hdfs/
# 也可以分成两步:
gzip /home/www/bigdata/target/fouryear.txt
hadoop fs -put /home/www/bigdata/target/fouryear.txt.gz /home/hdfs/

#!/usr/bin/env bash

for year in all/*
do
    echo -ne `basename $year .gz`"\t"
    gunzip -c $year | \
        awk '{ temp = substr($0, 88, 5) + 0;
            q = substr($0, 93, 1);
            if ( temp != 9999 && q ~/[01459]/ && temp > max) max = temp}
            END { print max }
            '
done

[root@localhost hadoop]# ./temp.sh
1901	317
1902	244

<?xml version="1.0" encoding="UTF-8"?>
<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>sunyu</groupId>
    <artifactId>hadoop</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <!-- Component versions are defined here -->
        <hadoop.version>2.9.1</hadoop.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

    </dependencies>

</project>

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MaxTempMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private static final int MISSING = 9999;

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 读取输入内容
        String line = value.toString();
        // 从内容中获取当前的年份
        String year = line.substring(15, 19);
        int airTemperature;
        // 读取当前记录对应的温度
        if (line.charAt(87) == '+') {
            airTemperature = Integer.parseInt(line.substring(88, 92));
        } else {
            airTemperature = Integer.parseInt(line.substring(87, 92));
        }
        String quality = line.substring(92, 93);
        // 输出年份和温度
        if (airTemperature != MISSING && quality.matches("[01459]")) {
            context.write(new Text(year), new IntWritable(airTemperature));
        }
    }
}

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.Text;

import java.io.IOException;

public class MaxTempReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int maxValue = Integer.MIN_VALUE;
        // 当前得到的输入是年份对应的温度列表,需要从中找到当前年份的最大温度
        for (IntWritable value : values) {
            maxValue = Math.max(maxValue, value.get());
        }
        context.write(key, new IntWritable(maxValue));
    }
}

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MaxTemp {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("用法: MaxTemp <输入路径> <输出路径>");
            System.exit(-1);
        }

        // 定义任务各参数
        Job job = new Job();
        job.setJarByClass(MaxTempMapper.class);
        job.setJobName("Max Temp");

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(MaxTempMapper.class);
        job.setReducerClass(MaxTempReduce.class);

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

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

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar MaxTemp all/sample.txt output

1901	317
1902	244

hdfs namenode -format

cd /usr/local/hadoop2.9.1/sbin
start-dfs.sh
start-yarn.sh

hadoop fs -mkdir -p /home/hdfs

hadoop fs -ls /
Found 1 items
drwxr-xr-x   - root supergroup          0 2018-11-18 15:49 /home

hadoop fs -ls -R /
drwxr-xr-x   - root supergroup          0 2018-11-18 15:49 /home
drwxr-xr-x   - root supergroup          0 2018-11-18 15:49 /home/hdfs

hadoop fs -put NOTICE.txt /home/hdfs/
hadoop fs -ls /home/hdfs
Found 1 items
-rw-r--r--   1 root supergroup      15915 2018-11-18 15:53 /home/hdfs/NOTICE.txt

hadoop fs -get /home/hdfs/NOTICE.txt /data/soft/

hadoop fs -rm /home/hdfs/NOTICE.txt

hadoop fs -rm -r /home/hdfs

<?xml version="1.0" encoding="UTF-8"?>
<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>sunyu</groupId>
    <artifactId>hadoop</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <!-- Component versions are defined here -->
        <hadoop.version>2.9.1</hadoop.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

    </dependencies>

</project>

package filesystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

public class RegexExcludePathFilter implements PathFilter {
  
  private final String regex;

  public RegexExcludePathFilter(String regex) {
    this.regex = regex;
  }

  public boolean accept(Path path) {
    return !path.toString().matches(regex);
  }
}

package filesystem;

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

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URL;

public class HadoopFileOper {
    static {
        // java 程序无法识别 hdfs 的路径,需要将 url 处理的工厂设置一下
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
    }

    /**
     * 通过 URL 形式读取文件.这种方式必须设置 UrlStreamHandlerFactory
     *
     * @throws Exception
     */
    public static void catFileByUrl() throws Exception {
        InputStream in = null;
        try {
            in = new URL("hdfs://localhost:9000/home/hdfs/NOTICE.txt").openStream();
            // 在输入流和输出流间复制数据
            IOUtils.copyBytes(in, System.out, 4096, false);
        } finally {
            // 关闭数据流
            IOUtils.closeStream(in);
        }
    }

    /**
     * 通过 FileSystem 读取 hdfs 中的文件
     *
     * @throws Exception
     */
    public static void catFileByFileSystem() throws Exception {
        // 读取配置文件中的内容,如 core-site.xml 中的配置
        Configuration conf = new Configuration();
        String uri = "hdfs://localhost:9000/home/hdfs/NOTICE.txt";
        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);
        } finally {
            IOUtils.closeStream(in);
        }
    }

    /**
     * 从本地文件系统上传文件到 hdfs
     * 上传文件时会自动创建目录
     *
     * @throws Exception
     */
    public static void uploadFile() throws Exception {
        String localSrc = "/home/www/bigdata/hadoop/test.txt";
        String dst = "hdfs://localhost:9000/home/hdfs/test.txt";

        InputStream in = new BufferedInputStream(new FileInputStream(localSrc));

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(dst), conf);
        OutputStream out = fs.create(new Path(dst), new Progressable() {
            public void progress() {
                System.out.print(".");
            }
        });

        IOUtils.copyBytes(in, out, 4096, true);
    }

    /**
     * 在 hdfs 上删除文件
     *
     * @throws Exception
     */
    public static void deleteFile() throws Exception {
        String toDel = "hdfs://localhost:9000/home/hdfs/test.txt";
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(toDel), conf);
        Path f = new Path(toDel);
        // 如果第一个参数是文件或空目录,第二个参数会被忽略
        // 如果第一个参数是非空的目录,且第二个值是 true,目录及其内容才会被删除
        boolean delete = fs.delete(f, true);
    }

    /**
     * 查看文件元数据
     *
     * @throws Exception
     */
    public static void listStatus() throws Exception {
        String uri = "hdfs://localhost:9000/home/hdfs/NOTICE.txt";
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);

        Path[] paths = new Path[1];
        paths[0] = new Path(uri);

        FileStatus[] status = fs.listStatus(paths);

        for (FileStatus stat : status) {
            System.out.printf(stat.getPath().toUri().getPath());
        }

        Path[] listedPaths = FileUtil.stat2Paths(status);
        for (Path p : listedPaths) {
            System.out.println(p);
        }
    }

    /**
     * 用表达式查找文件
     *
     * @throws Exception
     */
    public static void searchFile() throws Exception {
        String uri = "hdfs://localhost:9000/home/";
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        // 从第一个参数的路径中找文件,第二个参数是要过滤掉的文件模式
        FileStatus[] lists = fs.globStatus(new Path("/home/hdfs/*"), new RegexExcludePathFilter("^.*.doc$"));

        for (FileStatus item : lists) {
            System.out.printf(item.getPath().toUri().getPath());
        }
    }

    public static void main(String[] args) throws Exception {
//        catFileByUrl();
//        catFileByFileSystem();
//        uploadFile();
//        deleteFile();
//        listStatus();
        searchFile();
    }
}

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar filesystem/HadoopFileOper

// 复制文件
hadoop distcp file1 file2
// 复制目录,如果 dir2 不存在，会新建 dir2
// 如果 dir 存在，dir1 的内容会复制到 dir2 下，变为：dir2/dir1
// 如果只是想复制内容，可以使用 -overwrite 选项
hadoop distcp dir1 dir2

// -delete 会删除目标路径中和源路径不同的内容
// -p 会保留源文件的权限，块大小和复本数
hadoop distcp -update -delete -p hdfs://host1/dir hdfs://host2/dir2

hadoop distcp -update -delete -p webhdfs://host1:50070/dir webhdfs://host2:50070/dir2

package filesystem;

import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;

public class HadoopCompression {

    public static void streamCompression() throws Exception {
        DefaultCodec codec = new GzipCodec();
        Configuration conf = new Configuration();
        codec.setConf(conf);

        CompressionOutputStream out = codec.createOutputStream(System.out);
        IOUtils.copyBytes(System.in, out, 4096, false);
        out.finish();
    }

    public static void main(String[] args) throws Exception {
        streamCompression();
    }
}

mvn package

echo 'test stream' | hadoop jar target/hadoop-1.0-SNAPSHOT.jar filesystem/HadoopCompression | gunzip

gzip test.txt
hadoop fs -mkdir -p /home/hdfs
hadoop fs -put test.txt.gz /home/hdfs/testgzip.gz
hadoop fs -ls /home/hdfs
Found 1 items
-rw-r--r--   1 root supergroup        136 2018-12-04 14:41 /home/hdfs/testgzip.gz

package filesystem;

import java.io.InputStream;
import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;

import java.net.URI;

public class HadoopCompression {

    public static void fileCompression() throws Exception {
        String uri = "hdfs://localhost:9000/home/hdfs/testgzip.gz";
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);

        Path inputPath = new Path(uri);
        CompressionCodecFactory factory = new CompressionCodecFactory(conf);
        // 自动通过后缀名检测应该使用哪种压缩方式
        // 也可以强制指定某一种压缩方式
        // DefaultCodec codec = new GzipCodec();
        // codec.setConf(conf);
        CompressionCodec codec = factory.getCodec(inputPath);
        if (codec == null) {
            System.err.println("未检测到压缩方式: " + uri);
            System.exit(1);
        }

        String outputUri =
                CompressionCodecFactory.removeSuffix(uri, codec.getDefaultExtension());

        InputStream in = null;
        OutputStream out = null;
        try {
            in = codec.createInputStream(fs.open(inputPath));
            out = fs.create(new Path(outputUri));
            IOUtils.copyBytes(in, out, conf);
        } finally {
            IOUtils.closeStream(in);
            IOUtils.closeStream(out);
        }
    }

    public static void main(String[] args) throws Exception {
        fileCompression();
    }
}

mvn package
hadoop jar target/hadoop-1.0-SNAPSHOT.jar filesystem/HadoopCompression
18/12/04 14:42:34 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
18/12/04 14:42:34 INFO compress.CodecPool: Got brand-new decompressor [.gz]

hadoop fs -ls /home/hdfs
Found 2 items
-rw-r--r--   1 root supergroup      10240 2018-12-04 14:42 /home/hdfs/testgzip
-rw-r--r--   1 root supergroup        136 2018-12-04 14:41 /home/hdfs/testgzip.gz

hadoop fs -cat /home/hdfs/testgzip

public static void codecPool() throws Exception {
        Configuration conf = new Configuration();
        DefaultCodec codec = new GzipCodec();
        codec.setConf(conf);

        Compressor compressor = CodecPool.getCompressor(codec);
        CompressionOutputStream out = codec.createOutputStream(System.out, compressor);
        IOUtils.copyBytes(System.in, out, 4096, false);
        out.finish();
        CodecPool.returnCompressor(compressor);
    }

// 定义任务各参数
Configuration conf = new Configuration();
// Map 任务输出结果是否压缩
conf.setBoolean(Job.MAP_OUTPUT_COMPRESS, true);
// 设置输出结果压缩方式.这里使用 gzip
conf.setClass(Job.MAP_OUTPUT_COMPRESS_CODEC, GzipCodec.class,
        CompressionCodec.class);

class MyGenericWritable extends GenericWritable{
 
	//无参构造函数
	public MyGenericWritable() {
		
	}
	
	//有参构造函数
	public MyGenericWritable(Text text) {
		super.set(text);
	}
	
	//有参构造函数.如果当前是存的 Long 型
	public MyGenericWritable(LongWritable longWritable) {
		super.set(longWritable);
	}
 
	
	@Override
	protected Class<? extends Writable>[] getTypes() {
		// 提前定义好可能存在的类型
		return new Class[]{LongWritable.class,Text.class};
	}
	
}

package mapreduce;

import java.io.*;

import org.apache.hadoop.io.*;

public class IntPair implements WritableComparable<IntPair> {

  private int first;
  private int second;
  
  public IntPair() {
  }
  
  public IntPair(int first, int second) {
    set(first, second);
  }
  
  public void set(int first, int second) {
    this.first = first;
    this.second = second;
  }
  
  public int getFirst() {
    return first;
  }

  public int getSecond() {
    return second;
  }

  public void write(DataOutput out) throws IOException {
    out.writeInt(first);
    out.writeInt(second);
  }

  public void readFields(DataInput in) throws IOException {
    first = in.readInt();
    second = in.readInt();
  }
  
  @Override
  public int hashCode() {
    return first * 163 + second;
  }
  
  @Override
  public boolean equals(Object o) {
    if (o instanceof IntPair) {
      IntPair ip = (IntPair) o;
      return first == ip.first && second == ip.second;
    }
    return false;
  }

  @Override
  public String toString() {
    return first + "\t" + second;
  }
  
  public int compareTo(IntPair ip) {
    int cmp = compare(first, ip.first);
    if (cmp != 0) {
      return cmp;
    }
    return compare(second, ip.second);
  }
  
  /**
   * Convenience method for comparing two ints.
   */
  public static int compare(int a, int b) {
    return (a < b ? -1 : (a == b ? 0 : 1));
  }
  
}

package filesystem;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.ReflectionUtils;

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

public class SequenceFileOper {

    private static final String[] DATA = {
            "One, two, buckle my shoe",
            "Three, four, shut the door",
            "Five, six, pick up sticks",
            "Seven, eight, lay them straight",
            "Nine, ten, a big fat hen"
    };

    public static void writeFile() throws IOException {
        String uri = "hdfs://localhost:9000/home/hdfs/number.txt";
        Configuration conf = new Configuration();

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

        IntWritable key = new IntWritable();
        Text value = new Text();
        SequenceFile.Writer writer = null;

        try {
            writer = SequenceFile.createWriter(fs, conf, path,
                    key.getClass(), value.getClass());

            for (int i = 0; i < 100; i++) {
                key.set(100 - i);
                value.set(DATA[i % DATA.length]);
                System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key, value);
                // 每一行的内容是：key， value
                writer.append(key, value);
            }
        } finally {
            IOUtils.closeStream(writer);
        }
    }

    public static void readFile() throws IOException {
        String uri = "hdfs://localhost:9000/home/hdfs/number.txt";
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        Path path = new Path(uri);

        SequenceFile.Reader reader = null;
        try {
            reader = new SequenceFile.Reader(fs, path, conf);
            // 通过读取行内容确定键和值分别是哪种类型,从而通过反射得到对应的处理类
            Writable key = (Writable)
                    ReflectionUtils.newInstance(reader.getKeyClass(), conf);
            Writable value = (Writable)
                    ReflectionUtils.newInstance(reader.getValueClass(), conf);

            long position = reader.getPosition();
            // 顺序读取
            while (reader.next(key, value)) {
                String syncSeen = reader.syncSeen() ? "*" : "";
                System.out.printf("[%s%s]\t%s\t%s\n", position, syncSeen, key, value);
                position = reader.getPosition();
            }
        } finally {
            IOUtils.closeStream(reader);
        }
    }

    public static void main(String[] args) throws IOException {
        readFile();
    }
}

hadoop fs -text /home/hdfs/number.txt | head -n 10

String uri = "hdfs://localhost:9000/home/hdfs/mapnumber.txt";
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);

IntWritable key = new IntWritable();
Text value = new Text();
MapFile.Writer writer = null;
try {
  writer = new MapFile.Writer(conf, fs, uri,
      key.getClass(), value.getClass());
  
  for (int i = 0; i < 1024; i++) {
    key.set(i + 1);
    value.set(DATA[i % DATA.length]);
    writer.append(key, value);
  }
} finally {
  IOUtils.closeStream(writer);
}

<?xml version="1.0"?>
<configuration>
  <property>
    <name>size</name>
    <value>10</value>
    <description>Size</description>
  </property>
  
  <property>
    <name>weight</name>
    <value>heavy</value>
    <final>true</final>
    <description>Weight</description>
  </property>
  
  <property>
    <name>size-weight</name>
    <value>${size},${weight}</value>
    <description>Size and weight</description>
  </property>
</configuration>

Configuration conf = new Configuration();
conf.addResource("conf-1.xml");
String size = conf.get("color");
System.out.printf(size);

<property>
	<name>weight</name>
	<value>heavy</value>
	<final>true</final>
	<description>Weight</description>
</property>

hadoop fs -conf conf/hadoop-localhost.xml -ls /

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
	<property>
		<name>fs.defaultFS</name>
		<value>hdfs://localhost:9000</value>
		<description>NameNode</description>
	</property>
	<property>
    		<name>hadoop.tmp.dir</name>
    		<value>/usr/local/hadoop2.9.1/data</value>
  	</property>

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

	<property>
                <name>yarn.resourcemanager.hostname</name>
                <value>0.0.0.0</value>
        </property>
        <property>
                <name>yarn.nodemanager.aux-services</name>
                <value>mapreduce_shuffle</value>
        </property>

	<property>
        	<name>authname</name>
        	<value>sunyu</value>
        	<description>Size</description>
    	</property>
</configuration>

package mapreduce;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.*;

import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;

public class HdpConfig extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        for (Entry<String, String> entry : conf) {
            System.out.printf("%s=%s\n", entry.getKey(), entry.getValue());
        }
        return 0;
    }

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

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/HdpConfig -conf /usr/local/hadoop2.9.1/conf/hadoop-localhost.xml  | grep authname

authname=sunyu

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/HdpConfig -conf /usr/local/hadoop2.9.1/conf/hadoop-localhost.xml  -D authname=sun86yu

<?xml version="1.0" encoding="UTF-8"?>
<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>sunyu</groupId>
    <artifactId>hadoop</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <!-- Component versions are defined here -->
        <hadoop.version>2.9.1</hadoop.version>
        <mrunit.version>1.1.0</mrunit.version>
    </properties>
    <dependencies>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.9.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.mrunit</groupId>
            <artifactId>mrunit</artifactId>
            <version>${mrunit.version}</version>
            <classifier>hadoop2</classifier>
        </dependency>
    </dependencies>

</project>

package temptest;

import org.apache.hadoop.io.Text;

// 温度数据解析类
public class NcdcRecordParser {

    // 如果温度是 9999 表示温度数据已经丢失
    private static final int MISSING_TEMPERATURE = 9999;

    private String year;
    private int airTemperature;
    private boolean airTemperatureMalformed;
    private String quality;

    public void parse(String record) {
        year = record.substring(15, 19);
        airTemperatureMalformed = false;
        if (record.charAt(87) == '+') {
            airTemperature = Integer.parseInt(record.substring(88, 92));
        } else if (record.charAt(87) == '-') {
            airTemperature = Integer.parseInt(record.substring(87, 92));
        } else {
            airTemperatureMalformed = true;
        }
        quality = record.substring(92, 93);
    }

    public void parse(Text record) {
        parse(record.toString());
    }

    public boolean isValidTemperature() {
        return !airTemperatureMalformed && airTemperature != MISSING_TEMPERATURE
                && quality.matches("[01459]");
    }

    public boolean isMalformedTemperature() {
        return airTemperatureMalformed;
    }

    public String getYear() {
        return year;
    }

    public int getAirTemperature() {
        return airTemperature;
    }
}

package temptest;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MaxTemperatureMapper
        extends Mapper<LongWritable, Text, Text, IntWritable> {

    enum Temperature {
        MALFORMED
    }

    private NcdcRecordParser parser = new NcdcRecordParser();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        parser.parse(value);
        if (parser.isValidTemperature()) {
            int airTemperature = parser.getAirTemperature();
            context.write(new Text(parser.getYear()), new IntWritable(airTemperature));
        } else if (parser.isMalformedTemperature()) {
            System.err.println("忽略可能损坏的输入数据: " + value);
            context.getCounter(Temperature.MALFORMED).increment(1);
        }
    }
}

package temptest;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.Text;

import java.io.IOException;

public class MaxTemperatureReducer
        extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context)
            throws IOException, InterruptedException {

        int maxValue = Integer.MIN_VALUE;
        // 当前得到的输入是年份对应的温度列表,需要从中找到当前年份的最大温度
        for (IntWritable value : values) {
            maxValue = Math.max(maxValue, value.get());
        }
        context.write(key, new IntWritable(maxValue));
    }
}

package temptest;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;

import java.util.*;

import org.junit.Test;

import java.io.IOException;

public class HdpTest {

    @Test
    public void testMap() throws IOException, InterruptedException {
        Text value = new Text("0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N" +
                "0000001N9-00781+99999102001ADDGF108991999999999999999999");

        new MapDriver<LongWritable, Text, Text, IntWritable>()
                .withMapper(new MaxTemperatureMapper())
                .withInput(new LongWritable(0), value)
                .withOutput(new Text("1901"), new IntWritable(-78))
                .runTest();
    }

    @Test
    public void testReduce() throws IOException, InterruptedException {
        new ReduceDriver<Text, IntWritable, Text, IntWritable>()
                .withReducer(new MaxTemperatureReducer())
                .withInput(new Text("1901"),
                        Arrays.asList(new IntWritable(10), new IntWritable(5)))
                .withOutput(new Text("1901"), new IntWritable(10))
                .runTest();
    }
}

Process finished with exit code 0

java.lang.AssertionError: 1 Error(s): (Missing expected output (1901, -74) at position 0, got (1901, -78).)

package mapreduce;

import org.apache.hadoop.conf.Configured;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import temptest.MaxTemperatureReducer;
import temptest.MaxTemperatureMapper;

public class MaxTemperatureDriver extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.printf("使用方式: %s [控制参数] <输入数据路径> <结果输出路径>\n",
                    getClass().getSimpleName());
            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        Job job = new Job(getConf(), "Max temperature");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(MaxTemperatureMapper.class);
        // 设置 combiner 加速处理.节省带宽
        job.setCombinerClass(MaxTemperatureReducer.class);
        job.setReducerClass(MaxTemperatureReducer.class);

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

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

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

mvn package
hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/MaxTemperatureDriver /home/hdfs/fouryear.txt.gz /home/hdfs/output/maxtemp4/

[root@localhost hadoop]# hadoop fs -ls -R /home/hdfs/
-rw-r--r--   1 root supergroup     290515 2018-12-06 14:23 /home/hdfs/fouryear.txt.gz
drwxr-xr-x   - root supergroup          0 2018-12-06 14:25 /home/hdfs/output
drwxr-xr-x   - root supergroup          0 2018-12-06 14:25 /home/hdfs/output/maxtemp4
-rw-r--r--   1 root supergroup          0 2018-12-06 14:25 /home/hdfs/output/maxtemp4/_SUCCESS
-rw-r--r--   1 root supergroup         36 2018-12-06 14:25 /home/hdfs/output/maxtemp4/part-r-00000

[root@localhost hadoop]# hadoop fs -text /home/hdfs/output/maxtemp4/part-r-00000
1901	317
1902	244
1903	289
1904	256

package mapreduce;

import java.io.IOException;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import temptest.MaxTemperatureMapper;
import temptest.MaxTemperatureReducer;

public class HdpCounter extends Configured implements Tool {

    // 自定义计数器,包括两个值,用来统计系统中两种状态时的记录数
    enum SelfCounter {
        CNT_1901,
        CNT_1902,
        CNT_OTHER
    }

    static class TestMapperWithCounters
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        private NcdcRecordParser parser = new NcdcRecordParser();

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            parser.parse(value);

            int year = parser.getYearInt();

            if (year == 1901) {
                context.getCounter(SelfCounter.CNT_1901).increment(1);
            } else if (year == 1902) {
                context.getCounter(SelfCounter.CNT_1902).increment(1);
            } else {
                context.getCounter(SelfCounter.CNT_OTHER).increment(1);
            }

            int airTemperature = parser.getAirTemperature();
            context.write(new Text(parser.getYear()),
                    new IntWritable(airTemperature));
        }
    }

    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.printf("使用方式: %s [控制参数] <输入数据路径> <结果输出路径>\n",
                    getClass().getSimpleName());
            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        Job job = new Job(getConf(), "Max temperature");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(TestMapperWithCounters.class);
        // 设置 combiner 加速处理.节省带宽
        job.setCombinerClass(MaxTemperatureReducer.class);
        job.setReducerClass(MaxTemperatureReducer.class);

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

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

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

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/HdpCounter /home/hdfs/fouryear.txt.gz /home/hdfs/output/tempcounter/

Shuffle Errors
	BAD_ID=0
	CONNECTION=0
	IO_ERROR=0
	WRONG_LENGTH=0
	WRONG_MAP=0
	WRONG_REDUCE=0
mapreduce.HdpCounter$SelfCounter
	CNT_1901=6565
	CNT_1902=6565
	CNT_OTHER=13136
mapreduce.HdpCounter$TempStatus
	MISSING=44

mapred job -counter job_1544353902681_0001 'mapreduce.HdpCounter$TempStatus' MISSING

18/12/09 12:30:12 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
18/12/09 12:30:13 INFO mapred.ClientServiceDelegate: Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
44

mapred job -counter job_1544353902681_0001 'org.apache.hadoop.mapreduce.TaskCounter' MAP_INPUT_RECORDS

org.apache.hadoop.mapreduce.TaskCounter;
org.apache.hadoop.mapreduce.FileSystemCounter;
org.apache.hadoop.mapreduce.JobCounter;
org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter;

<property>  
        <name>mapreduce.jobhistory.address</name>  
        <value>0.0.0.0:10020</value>  
</property>

/usr/local/hadoop2.9.1/sbin/mr-jobhistory-daemon.sh start historyserver

package mapreduce;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.util.*;

public class CounterFields extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        if (args.length != 1) {
            JobBuilder.printUsage(this, "<job ID>");
            return -1;
        }
        String jobID = args[0];
        Cluster cluster = new Cluster(getConf());
        Job job = cluster.getJob(JobID.forName(jobID));
        if (job == null) {
            System.err.printf("任务 ID %s 未找到.\n", jobID);
            return -1;
        }
        if (!job.isComplete()) {
            System.err.printf("任务 %s 还未完成.\n", jobID);
            return -1;
        }

        Counters counters = job.getCounters();
        long total = counters.findCounter(HdpCounter.SelfCounter.CNT_1901).getValue();

        System.out.printf("1901年的数据条数: %d\n", total);
        return 0;
    }

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

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/MissingTemperatureFields job_1544353902681_0001

18/12/09 12:58:42 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
18/12/09 12:58:43 INFO mapred.ClientServiceDelegate: Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
1901年的数据条数: 6565

package mapreduce;

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.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class SortData extends Configured implements Tool {

    static class TransMap
            extends Mapper<LongWritable, Text, IntWritable, Text> {

        private NcdcRecordParser parser = new NcdcRecordParser();

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            parser.parse(value);

            int airTemperature = parser.getAirTemperature();

            context.write(new IntWritable(airTemperature), value);
        }
    }

    public int run(String[] args) throws Exception {
        Job job = new Job(getConf(), "Sort Data");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(TransMap.class);

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

        // 不需要 reduce
        job.setNumReduceTasks(0);
        // 输出为顺序文件
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        // 开启压缩
        SequenceFileOutputFormat.setCompressOutput(job, true);
        // Gzip 压缩
        SequenceFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        // 块压缩
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);

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

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

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/SortData /home/hdfs/fouryear.txt.gz /home/hdfs/output/sequenfile/

hadoop fs -text /home/hdfs/output/sequenfile/part-m-00000 | head -n 5

-78	0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N0000001N9-00781+99999102001ADDGF108991999999999999999999
-72	0029029070999991901010113004+64333+023450FM-12+000599999V0202901N008219999999N0000001N9-00721+99999102001ADDGF104991999999999999999999
-94	0029029070999991901010120004+64333+023450FM-12+000599999V0209991C000019999999N0000001N9-00941+99999102001ADDGF108991999999999999999999
-61	0029029070999991901010206004+64333+023450FM-12+000599999V0201801N008219999999N0000001N9-00611+99999101831ADDGF108991999999999999999999
-56	0029029070999991901010213004+64333+023450FM-12+000599999V0201801N009819999999N0000001N9-00561+99999101761ADDGF108991999999999999999999

package mapreduce;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SortSequeneByMR extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        Job job = new Job(getConf(), "Sort Data");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setInputFormatClass(SequenceFileInputFormat.class);

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

        // reduce 个数
        job.setNumReduceTasks(10);
        // 输出为顺序文件
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        // 开启压缩
        SequenceFileOutputFormat.setCompressOutput(job, true);
        // Gzip 压缩
        SequenceFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        // 块压缩
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);

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

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

hadoop fs -ls -R /home/hdfs/output/sortsequen
-rw-r--r--   1 root supergroup          0 2018-12-10 09:01 /home/hdfs/output/sortsequen/_SUCCESS
-rw-r--r--   1 root supergroup      42828 2018-12-10 09:00 /home/hdfs/output/sortsequen/part-r-00000
-rw-r--r--   1 root supergroup      37771 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00001
-rw-r--r--   1 root supergroup      42849 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00002
-rw-r--r--   1 root supergroup      22710 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00003
-rw-r--r--   1 root supergroup      29258 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00004
-rw-r--r--   1 root supergroup      11462 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00005
-rw-r--r--   1 root supergroup      41244 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00006
-rw-r--r--   1 root supergroup      40074 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00007
-rw-r--r--   1 root supergroup      33187 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00008
-rw-r--r--   1 root supergroup      24110 2018-12-10 09:01 /home/hdfs/output/sortsequen/part-r-00009

package mapreduce;

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.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

public class TotalSortBySample extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        Job job = new Job(getConf(), "Sort Data");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setInputFormatClass(SequenceFileInputFormat.class);

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

        // reduce
        job.setNumReduceTasks(10);
        // 输出为顺序文件
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        // 开启压缩
        SequenceFileOutputFormat.setCompressOutput(job, true);
        // Gzip 压缩
        SequenceFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        // 块压缩
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);

        // Hadoop 自动抽样并分区
        // 使用全排序的分区
        job.setPartitionerClass(TotalOrderPartitioner.class);
        // 使用随机抽样进行分区.采样率为 10%，最多采样 10000 条，最多划分 10 个分区
        InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<IntWritable, Text>(0.1, 10000, 10);

        InputSampler.writePartitionFile(job, sampler);

        // 将分区配置添加到分布式缓存
        Configuration conf = job.getConfiguration();
        String partitionFile = TotalOrderPartitioner.getPartitionFile(conf);
        URI partitionUri = new URI(partitionFile);
        job.addCacheFile(partitionUri);

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

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

mvn package

hadoop jar target/hadoop-1.0-SNAPSHOT.jar mapreduce/TotalSortBySample /home/hdfs/output/sequenfile/part-m-00000 /home/hdfs/output/totalsort

hadoop fs -ls /home/hdfs/output/totalsort
Found 11 items
-rw-r--r--   1 root supergroup          0 2018-12-10 09:48 /home/hdfs/output/totalsort/_SUCCESS
-rw-r--r--   1 root supergroup      27593 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00000
-rw-r--r--   1 root supergroup      34043 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00001
-rw-r--r--   1 root supergroup      25292 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00002
-rw-r--r--   1 root supergroup      36274 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00003
-rw-r--r--   1 root supergroup      28948 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00004
-rw-r--r--   1 root supergroup      37523 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00005
-rw-r--r--   1 root supergroup      32046 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00006
-rw-r--r--   1 root supergroup      33910 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00007
-rw-r--r--   1 root supergroup      32300 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00008
-rw-r--r--   1 root supergroup      34734 2018-12-10 09:48 /home/hdfs/output/totalsort/part-r-00009

hadoop fs -text /home/hdfs/output/totalsort/part-r-00000 | tail -n 3
18/12/10 09:49:27 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
18/12/10 09:49:27 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:49:27 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:49:27 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:49:27 INFO compress.CodecPool: Got brand-new decompressor [.gz]

-94	0029029720999991903012120004+60450+022267FM-12+001499999V0209991C000019999999N0000001N9-00941+99999102601ADDGF100991999999999999999999
-94	0029029070999991902020620004+64333+023450FM-12+000599999V0201401N006219999999N0000001N9-00941+99999100661ADDGF108991999999999999999999
-94	0029029600999991904120906004+61183+022617FM-12+005199999V0201601N002119999999N0000001N9-00941+99999099311ADDGF108991999999999999999999


hadoop fs -text /home/hdfs/output/totalsort/part-r-00001 | head -n 3
18/12/10 09:52:02 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
18/12/10 09:52:02 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:52:02 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:52:02 INFO compress.CodecPool: Got brand-new decompressor [.gz]
18/12/10 09:52:02 INFO compress.CodecPool: Got brand-new decompressor [.gz]
-89	0029029810999991901021213004+59500+020350FM-12+002699999V0200201N008719999999N0000001N9-00891+99999100841ADDGF104991999999999999999999
-89	0029029500999991902122920004+61483+021350FM-12+000699999V0200701N009819999999N0000001N9-00891+99999098421ADDGF108991999999999999999999
-89	0029029500999991902031113004+61483+021350FM-12+000699999V0200501N004119999999N0000001N9-00891+99999100991ADDGF100991999999999999999999

package mapreduce;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
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.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class AssistSort extends Configured implements Tool {

    static class MaxTemperatureMapper
            extends Mapper<LongWritable, Text, IntPair, NullWritable> {

        private NcdcRecordParser parser = new NcdcRecordParser();

        @Override
        protected void map(LongWritable key, Text value,
                           Context context) throws IOException, InterruptedException {

            parser.parse(value);
            if (parser.isValidTemperature()) {
                context.write(new IntPair(parser.getYearInt(),
                        parser.getAirTemperature()), NullWritable.get());
            }
        }
    }

    static class MaxTemperatureReducer
            extends Reducer<IntPair, NullWritable, IntPair, NullWritable> {

        @Override
        protected void reduce(IntPair key, Iterable<NullWritable> values,
                              Context context) throws IOException, InterruptedException {

            context.write(key, NullWritable.get());
        }
    }

    public static class FirstPartitioner
            extends Partitioner<IntPair, NullWritable> {

        @Override
        public int getPartition(IntPair key, NullWritable value, int numPartitions) {
            return Math.abs(key.getFirst() * 127) % numPartitions;
        }
    }

    public static class KeyComparator extends WritableComparator {
        protected KeyComparator() {
            super(IntPair.class, true);
        }

        @Override
        public int compare(WritableComparable w1, WritableComparable w2) {
            IntPair ip1 = (IntPair) w1;
            IntPair ip2 = (IntPair) w2;
            int cmp = IntPair.compare(ip1.getFirst(), ip2.getFirst());
            if (cmp != 0) {
                return cmp;
            }
            return -IntPair.compare(ip1.getSecond(), ip2.getSecond());
        }
    }

    public static class GroupComparator extends WritableComparator {
        protected GroupComparator() {
            super(IntPair.class, true);
        }

        @Override
        public int compare(WritableComparable w1, WritableComparable w2) {
            IntPair ip1 = (IntPair) w1;
            IntPair ip2 = (IntPair) w2;
            return IntPair.compare(ip1.getFirst(), ip2.getFirst());
        }
    }

    public int run(String[] args) throws Exception {
        Job job = new Job(getConf(), "Assistant Sort Data");
        job.setJarByClass(getClass());

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(MaxTemperatureMapper.class);
        job.setPartitionerClass(FirstPartitioner.class);
        job.setSortComparatorClass(KeyComparator.class);
        job.setGroupingComparatorClass(GroupComparator.class);
        job.setReducerClass(MaxTemperatureReducer.class);
        job.setOutputKeyClass(IntPair.class);
        job.setOutputValueClass(NullWritable.class);

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

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

file:///Users/sunyu/ebook/%E4%B8%AA%E4%BA%BA%E6%80%BB%E7%BB%93/%E5%A4%A7%E6%95%B0%E6%8D%AE/storm/storm.html#flume-hdfs-mysql-storm-%E6%B5%81%E9%87%8F%E5%AE%9E%E6%97%B6%E5%88%86%E6%9E%90

file:///Users/sunyu/ebook/%E4%B8%AA%E4%BA%BA%E6%80%BB%E7%BB%93/%E5%A4%A7%E6%95%B0%E6%8D%AE/storm/storm.html#nginx-kafka-hbase-storm-%E5%B9%BF%E5%91%8A%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90

# 日志汇总后要存放在HDFS的目录, 这里有日期命名
master.sinks.log-sink-master.hdfs.path = hdfs://localhost:9000/flume/data/%Y-%m-%d

对一些应用程序而言，事件写入HDFS时使用一个时间戳就够了，但有些应用有多层 Flume 代理，那么创建事件和存入的时间可能就有明显的差异，特别是代理出现停机的情况下。这时候我们可以对 HDFS 的 sink 的 ```hdfs.useLocalTimeStap```属性进行设置，以便使用由运行 HDFS 的 Flume 代理产生时间戳。

文件格式
---
一般而言，使用二进制格式来存储数据更好，因为它生成的文件比使用文本格式的文件更小。 HDFS sink 使用的文件格式由 ```hdfs.fileType``` 及其他一些属性的组合控制。

默认的 ```hdfs.fileType``` 文件格式是 ```SequenceFile```，也就是把事件写入一个顺序文件，这个文件中，键就是事件的时间戳，是一个 LongWritable 类型。值的类型是 BytesWritable。如果 ```hdfs.writeFormat```属性被设置为 Text，那么值的类型就是 Text 类型，不是 BytesWritable 了。

同时，还可以为 sink 设置压缩方式。如：

扇出
---

可以从一个 source 获取数据，分发到之个 sink ，当然是通过之个 channle。这样就可以实现这种业务：数据存入 Hbase 持久化，同时存入 Solr 提供实时搜索。

多个 sink 配置可能如下：

复用
---
还有一种用法是在 source 上设置一个复用选择器。这样就可以让一部分事件到一个 sink,一部分到另一个。用法:

代理层分发
---
通常，我们是让代理接收事件，然后写入 sink。但如果代理有很多，同时写 sink 时，压力会比较大。这时候可以引入一层代理分发层。让代理把数据先传到分发层，分发代理数目远小于节点数。这就要求代理和分发层之前有一个特殊的数据传递。如：

Sink组
---
sink 组可以让多个 sink 当作一个 sink 处理，以实现故障转移和负载均衡。

HBase
===

Hbase 是建立在 HDFS 上的一个面向列的分布式数据库。它和 RDBMS 相似，但它有单元格的版本，每行数据是排序的，而且只要列族预告存在，可以随时添加新列。

HBase 依赖 Zookeeper。用来协调集群。

Hbase 通过在 HDFS 上提供随机读/写来解决 Hadoop 不能处理的问题。Hbase 从底层设计就开始考虑各种可扩展问题。一张表可以很大，数十亿个行，数百万个列；水平分区可以在上千个普通的商用机节点上。这样保证了高效的存储，搜索功能。

RDBMS 的模式比较固定，面向行，且具有 ACID和复杂的SQL查询引擎。强调事务的强一致性。在RDBMS中可以很容易建立二级索引，执行复杂的连接，排序，分组等操作，对或、行、列中的数据进行分页存放。

在小型应用中，诸如 MySQL 这样的 RDMBS 很灵活，方便。但数据规模和并发读写这两方面中任何一个如果要向上大规模的扩展，RDMBS 就会碰到瓶颈。通常会需要我们放松 ACID，让数据、业务管理更复杂。

HBase 有一些特性：

1. 没有真正的索引。行是顺序存储的，每行中的列也是，所以不存在索引膨胀的问题，而且插入性能和表的大小无关。
2. 自动分区。在表增长时，表会自动分裂成区域，并分布到可用的节点上。
3. 线性扩展。当新增节点时，把它指向现有集群并运行，系统会自动平衡，均衡分布。
4. 普通硬件支持。集群中的硬件不需要是很高配置的。普通的商用硬件就行。
5. 容错性。大量节点意味着每个节点的重要性并不突出。不用担心单点的问题。
6. 批处理。MapReduce的功能可以让我们用全并行的分布式作业处理业务。

ZooKeeper
===


分布式应用中，主要的困难是会出现“部分失败”。两个节点间通信时，如果出现网络错误，发送者无法知道接收者是否已经收到。zookeeper 是为了解决这一问题而诞生的，它不能解决避免出现失败，也不会隐藏失败，它提供一组工具，使得出现部分失败时能够妥善的处理。

它的特点有：

1. 设计简单。zookeeper 的核心是一个精简的文件系统，它提供一些简单的操作和一些额外的抽象操作，如排序，通知。
2. 高可用。它可以运行于一组机器之上，并且设计上有高可用性。可以帮助系统避免出现单点故障。
3. 松耦合。在交互的过程中，参与者只需要关注自己的信息，不用管它方。就算自己处理过程结束了，其它进程还可以读取到这条信息。
4. 高性能。它的吞量可以超过每秒 10000 个操作。

安装
---

进入下载页面：http://mirror.bit.edu.cn/apache/zookeeper/ 选择一个稳定版本。

设置简单的配置文件 conf/zoo.cfg:

***单机模式:***

启动服务：

启动后可以看到 2181 端口已经被监听。这时候可以做一些简单的测试：

imok 的意思是 zookeeper 服务器返回：i'm ok。还有一些其它命令可以用来管理 zookeeper，都是类似这种四个字母的组合。如：

| 命令 | 描述 |
|-----|-----|
| ruok | 如果服务器正在运行且未处于出错状态，返回 imok |
| conf | 输出服务器的配置信息,根据 zoo.cfg |
| envi | 输出服务器的环境变量，包括版本，JAVA版本及其它信息 |
| srvr | 服务器的统计信息，包括延迟统计，znode 数量，运行模式等 |
| stat | 服务器的统计信息和已连接的客户端 |
| srst | 重置服务器的统计信息 |
| isro | 显示服务器是否处于只读模式或者读写模式 |
| dump | 列出集合体中所有会话和临时 znode。必须连接到 leader 上才能用该命令 |
| cons | 列出所有服务器客户端的连接统计信息 |
| crst | 重置连接统计信息 |
| wchs | 列出服务器上所有监听的摘要信息 |
| wchc | 按连接列出服务器上所有的监听 |
| wchp | 按 znode 路径列出服务器上所有的监听 |
| mntr | 按JAVA属性格式列出服务器统计信息 |

应用
---
使用 zookeeper 最常用的就是在上面创建节点，设置节点的值。以及节点内容编辑，删除。代码可能如下：

***ZkpTest.java***

/**
 * 创建组
 *
 * @param groupName 组名
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public void create(String groupName) throws IOException, KeeperException,
        InterruptedException {
    ZooKeeper zk = new ZooKeeper("192.168.111.111", SESSION_TIMEOUT, this);

    String path = "/" + groupName;
    // 创建持久化节点,不是临时的.节点的内容为空,节点访问权限是完全开放
    String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);
    System.out.println("创建组 " + createdPath);
}

/**
 * 在组里创新子节点
 *
 * @param groupName
 * @param memberName
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public void join(String groupName, String memberName) throws IOException, KeeperException,
        InterruptedException {
    ZooKeeper zk = new ZooKeeper("192.168.111.111", SESSION_TIMEOUT, this);

    connectedSignal.await();

    String path = "/" + groupName + "/" + memberName;
    // 创建临时节点
    String createdPath = zk.create(path, null/*data*/, Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL);
    System.out.println("创建节点 " + createdPath);
}

/***
 * 列出组里所有的节点
 * @param groupName
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public void list(String groupName) throws IOException, KeeperException,
        InterruptedException {
    String path = "/" + groupName;

    try {
        ZooKeeper zk = new ZooKeeper("192.168.111.111", SESSION_TIMEOUT, this);

        List<String> children = zk.getChildren(path, false);
        if (children.isEmpty()) {
            System.out.printf("组里无节点 %s\n", groupName);
            System.exit(1);
        }
        for (String child : children) {
            System.out.println(child);
        }
    } catch (KeeperException.NoNodeException e) {
        System.out.printf("组 %s 不存在\n", groupName);
        System.exit(1);
    }
}

/**
 * 删除组以及组里的子节点
 * @param groupName
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public void delete(String groupName) throws IOException, KeeperException,
        InterruptedException {
    String path = "/" + groupName;

    try {
        ZooKeeper zk = new ZooKeeper("192.168.111.111", SESSION_TIMEOUT, this);

        List<String> children = zk.getChildren(path, false);
        for (String child : children) {
            zk.delete(path + "/" + child, -1);
            System.out.printf("删除 %s 组里的节点 %s\n", groupName, child);
        }
        zk.delete(path, -1);
    } catch (KeeperException.NoNodeException e) {
        System.out.printf("组 %s 不存在\n", groupName);
        System.exit(1);
    }
}

public void process(WatchedEvent event) {
    if (event.getState() == KeeperState.SyncConnected) {
        connectedSignal.countDown();
    }
}

public static void main(String[] args) throws Exception {
    ZkpTest test = new ZkpTest();
    test.create("test");
    test.list("test");
    test.join("test", "anode");
    test.list("test");
    test.delete("test");
    test.list("test");
} }

详细说明
---

***数据模型***

Zookeeper 维护着一个类似文件系统的树形层次结构。树中的节点叫 znode。它可以用来存储数据，并且有访问权限。这种设计可以用来实现协调服务，而不是大数据的存储，通常一个 znode 存储的数据都被限制在 1MB 以内。

zookeeper 的数据访问具有原子性。客户端在读取一个 znode 的数据时，要么读到所有数据，要么读取失败，不会只读到部分数据。同样，写操作也将替换所有 znode 里的数据，不会部分节点成功写入部分失败。zookeeper 不支持添加操作。

zookeeper 中的路径象文件系统的路径。使用斜杠分隔目录。但 zookeeper 的路径必须是绝对路径，没有相对路径的概念。所有的路径表示必须是规范的，每条路径只有唯一的一种表示方法，不支持类似 /a/./b 这种方式。在文件系统中 . 可以表示当前目录，而 zookeeper 中不支持。

>znode 有两种类型，一种临时的，一种持久的。类型在 znode 创建后就不会再变更。临时节点在客户端会话结束时就会被删除。持久节点只有显式的调用删除功能才会删除。

创建节点时，还可以指定序号标识。这样，znode 名称后就会附加一个序号，该值是一个单调递增的计数器。由它的父节点维护。如，创建一个名为 /test- 的顺序节点，它的名称可能是 /test-1, /test-2 这样。顺序号可以用来为并发时的事件排序，这样客户端就能通过序号推断事件发生的先后。以及实现分布式锁。

我们还可以在节点上添加监听，当节点发生变化时通知监听的客户端。监听是一次性的，事件触发后就消失了，所以通常是在触发后再添加新的监听。

***操作***

zookeeper 中有 9 种操作：

| 操作 | 说明 |
| ----| ---- |
| create | 创建一个 znode，必须要有父节点 |
| delete | 删除一个 znode，它不能有子节点 |
| exists | 检测节点是否存在并查询它的元数据 |
| getACL, setACL | 获取或设置一个节点的访问权限 |
| getChildren | 获取一个节点的子节点列表 |
| getData, setData | 获取或设置一个 znode 保存的数据 |
| sync | 将客户端 znode 数据和 zookeeper 同步 |

zookeeper 中更新数据是有条件的。在使用 delete 或 setData 时，必须提供被更新节点的版本号（可通过 exists 获取）。如果版本号不匹配，更新操作会失败。

zookeeper 中还有一个被称为 multi 的操作，它是将多个基本操作作为一个单元，并保证这些操作同时被成功执行。类似事务。该特性可以用于需要保持全局一致性的数据结构。

create, delete, setData 这些操作上可以触发设置在 exists, getChildren, getData 上的监听功能。权限相关设置不行。

1. 当被监听的 znode 被创建、删除、或数据被更新时，设置在 exists 上的监听会被触发。
2. 当被监听的 znode 被删除或数据被更新时，设置在 getData 操作上的监听被触发。创建节点不会触发。
3. 被监听的 znode 被创建或删除时，或所观察的 znode 自己被删除时，设置在 getChildren 上的观察将被触发。

监听节点事件触发对应关系是：

| 监听操作 | 创建 znode | 创建子节点 | 删除 znode | 删除子节点 | setData |
| ---- | ---- | ---- | ---- | ---- | ---- |
| exists | 触发 | | 触发 | | 触发 |
| getData | | | 触发 | | 触发 |
| getChildren | | 触发 | 触发 | 触发 | |


每个 znode 创建时都会带有一个 ACL 列表，用于决定谁可以对它执行何种操作。ACL 依赖于 zookeeper 客户端的身份验证机制 。有三种身份验证机制：

1. digest, 通过用户名和密码来识别
2. sasl, 通过 Kerberos 识别
3. ip，通过客户端的IP地址来识别

在建立一个会话后，客户端可以对自己进行身份验证。如：

设置权限可以如下：