hadoop_tools: Head.java

/*
 * Copyright 1997-2012
 * http://www.pconline.com.cn
 */
package cn.pconline.hadooptools;

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.commons.compress.archivers.ArchiveException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 *
 * @author larry <larryliao@pconline.com.cn>
 */
public class Head {

    private static final Log LOG =
            LogFactory.getLog(Head.class.getName());
    static final String NAME = "Head";
    static int MaxLineNumber = 10;

    public static void main(String[] args) throws IOException, ArchiveException {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

        if (otherArgs.length < 1) {
            System.err.println("ERROR: Wrong number of parameters: " + args.length);
            System.err.println("Usage: Head -conf <configuration file> -D <property=value>"
                    + " <filePath> [lineNumber]");

            System.exit(-1);
        }

        FileSystem hdfs = FileSystem.get(conf); //获埗HDFS文件系统的对象

        Path inputDir = new Path(otherArgs[0]);
        if (otherArgs.length > 1) {
            MaxLineNumber = Integer.parseInt(otherArgs[1]) > 0 ? Integer.parseInt(otherArgs[1]) : 10;
        }

        byte[] buffer = new byte[4096];
        FSDataInputStream inputStream = hdfs.open(inputDir);

        GzipCodec codec = new GzipCodec();
        codec.setConf(conf);
        CompressionInputStream input = codec.createInputStream(inputStream);

        int lineNumber = 0;
        String lastLine = null;
        input.read(buffer);
        while (lineNumber < MaxLineNumber && buffer != null) {
            String[] out = new String(buffer, "UTF-8").split("\n");
            for (int i = 0; i < out.length; i++) {
                if (i == out.length - 1) {
                    lastLine = out[i];
                } else {
                    if (i == 0 && lastLine != null) {
                        System.out.println(lastLine + out[i]);
                    } else {
                        System.out.println(out[i]);
                    }
                    lineNumber++;
                }

                if (lineNumber >= MaxLineNumber) {
                    break;
                }
            }
            if (lineNumber < MaxLineNumber) {
                input.read(buffer);
            }
        }

        hdfs.close();
    }
}