backupjava

import java.io.IOException;
import java.util.ArrayList ;
import java.util.Collections ;
import java.util.Comparator ;
import java.util.Date ;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ContentSummary ;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;

/**
* Copyright 2009 Rapleaf, Inc.
* Cesar Delgado
* cesar AT rapleaf.com
* License: Apache License 2.0
**/


public class Backup {

	/**
	* Prints out usage
	**/
	static void usage() {
		System.err.println("Usage: hadoop Backup <path on hdfs> <path on local fs> <unix time> [<max size to backup in bytes>]") ;
		System.exit(1) ;
	}


	public static void main (String[] argv) throws IOException {

		if (argv.length < 3)
			usage() ;

		Path baseDir = new Path(argv[0]) ; 	// HDFS path
		String localPath = argv[1] ; 		// Local path
		long minDate = Long.parseLong(argv[2]) ;// UNIX date since epoch of last backup
		long maxDate = new Date().getTime() / 1000 ;	// UNIX date for right now
		long tmpDate = 0 ;
		long size = 0 ;
		if (argv.length == 4)
			size = Long.parseLong(argv[3]) ; //Get the max size param

		Backup bak = new Backup() ;
		ArrayList<Path> pathList = new ArrayList<Path>(2000) ;

		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);

		if (fs.getFileStatus(baseDir).isDir()) { //If the HDFS path is a dir continue
			System.err.println("Backing up: " + baseDir.toUri().getPath()) ;
			bak.checkDir(fs, minDate, maxDate, baseDir, pathList) ;
			System.err.println("Number of files searched = " + pathList.size()) ;

			Collections.sort(pathList, new DateComparator(conf, fs)) ;

			tmpDate = bak.backupFiles( localPath, fs, pathList, size) ;
		}

		if (tmpDate == 0) { // If not size limit reached print out date for right now
			System.out.println(maxDate) ;
		} else { // Print out date for last file backed up
			System.err.println("Size limit reached.") ;
			System.out.println(tmpDate) ;
		}
	}

	static class DateComparator implements Comparator {

		Configuration conf ;
                FileSystem fs ;

		public DateComparator(Configuration c, FileSystem f) {
			conf = c ;
			fs = f ;
		}

		public int compare(Object path1, Object path2) {

			try {
				long date1, date2 ;
				date1 = fs.getFileStatus((Path) path1).getModificationTime() ;
				date2 = fs.getFileStatus((Path) path2).getModificationTime() ;

				if( date1 > date2 )
					return 1;
				else if(date1 < date2 )
					return -1;
				else
					return 0;
			} catch (IOException e) {
				System.err.println("Something went wrong when trying to compare dates") ;
				System.err.println(e) ;
				System.exit(1) ;
				return 0 ;
			}

		}
	}

	public Backup() {
		
	}

	/**
	* Method to move files from HDFS to local filesystem
	*
	* localPath: 	Path on the machines filesystem
	* fs:		FileSystem object from HDFS
	* pathList:	List of paths for files that might need to be backed up
	* size:		max size in bytes to be backed up
	*
	* Returns	Date of the last files backed up if reached size limit, else, zero
	**/

	public long backupFiles(String localPath, FileSystem fs, ArrayList<Path> pathList, long size) {
		Path fsPath ;
		long tmpSize = 0 ;
		long tmpDate = 0 ;

		// Start iterating over all paths
		for (Path hdfsPath : pathList) {
			try {
				tmpSize = tmpSize + fs.getContentSummary(hdfsPath).getLength() ;
				if ((tmpSize <= size) || (size == 0)) {
					tmpDate = fs.getFileStatus(hdfsPath).getModificationTime() / 1000 ;
					System.err.print("File: " + hdfsPath.toUri().getPath()) ;

					fsPath = new Path(localPath + hdfsPath.toUri().getPath()) ;
					fs.copyToLocalFile(hdfsPath, fsPath) ;
				} else {
					return tmpDate ;
				}

				System.err.println(" Done") ;
			} catch (IOException e) {
				System.err.println("Something wrong with the file") ;
				System.err.println(e) ;
				System.exit(1) ;
				return 0 ;
			}
		}
		return 0 ;
	}

	/**
	* Method to go though the HDFS filesystem in a DFS to find all files
	*
	* fs:		FileSystem object from HDFS
	* minDate:      Oldest date for files to be backed up
	* maxDate:	Newest date for files to be backed up
	* p:		Path in HDFS to look for files
	* pathList:	Will be filled with all files in p
	**/

	public void checkDir(FileSystem fs, long minDate, long maxDate, Path p, ArrayList<Path> pathList) {
		long tmpDate ;
		FileStatus[] fStat ;
		try {
			if (fs.getFileStatus(p).isDir()) { // If this is a directory
				fStat = fs.listStatus(p) ; 
				for (int i = 0; i < fStat.length; i++) { // Do a recursive call to all elements
					checkDir(fs, minDate, maxDate, fStat[i].getPath(), pathList) ;
				}
			} else { // If not a directory then we've found a file
				tmpDate = fs.getFileStatus(p).getModificationTime() / 1000 ;
				if ((minDate <= tmpDate) && (maxDate > tmpDate)){
					pathList.add(p) ;
				}
			}
		} catch (IOException e) {
			System.err.println("Could not open" + p) ;
			System.err.println(e) ;
			System.exit(1) ;
		}
	}

}