core/src/main/java/org/apache/hadoop/chukwa/extraction/demux/MoveToRepository.java - chukwa - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.chukwa.extraction.demux;


 import java.io.IOException;
 import java.net.URI;
 import java.text.SimpleDateFormat;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;

 import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
 import org.apache.hadoop.chukwa.util.HierarchyDataType;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.log4j.Logger;

 // TODO
 // First version of the Spill
 // need some polishing

 public class MoveToRepository {
   static Logger log = Logger.getLogger(MoveToRepository.class);

   static ChukwaConfiguration conf = null;
   static FileSystem fs = null;
   static SimpleDateFormat sdf = new java.text.SimpleDateFormat("yyyyMMdd");
   static Calendar calendar = Calendar.getInstance();

   static Collection<Path> processClusterDirectory(Path srcDir, String destDir)
       throws Exception {
     log.info("processClusterDirectory (" + srcDir.getName() + "," + destDir
         + ")");
     FileStatus fstat = fs.getFileStatus(srcDir);
     Collection<Path> destFiles = new HashSet<Path>();

     if (!fstat.isDir()) {
       throw new IOException(srcDir + " is not a directory!");
     } else {
       FileStatus[] datasourceDirectories = fs.listStatus(srcDir);

       for (FileStatus datasourceDirectory : datasourceDirectories) {
         log.info(datasourceDirectory.getPath() + " isDir?"
             + datasourceDirectory.isDir());
         if (!datasourceDirectory.isDir()) {
           throw new IOException(
               "Top level datasource directory should be a directory :"
                   + datasourceDirectory.getPath());
         }

         PathFilter filter = new PathFilter()
         {public boolean accept(Path file) {
           return file.getName().endsWith(".evt");
         }  };
         //CHUKWA-648:  Make Chukwa Reduce Type to support hierarchy format
         // to processDataSourceDirectory according to hierarchy data type format
         List<FileStatus> eventfiles = HierarchyDataType.globStatus(fs, datasourceDirectory.getPath(),filter,true);
         for (FileStatus eventfile : eventfiles){
           Path datatypeDir = eventfile.getPath().getParent();
           String dirName = HierarchyDataType.getDataType(datatypeDir, srcDir);

           Path destPath = new Path(destDir + "/" + dirName);
           log.info("dest directory path: " + destPath);
           log.info("processClusterDirectory processing Datasource: (" + dirName
               + ")");
           StringBuilder dtDir = new StringBuilder(srcDir.toString()).append("/").append(dirName);
           log.debug("srcDir: " + dtDir.toString());
           processDatasourceDirectory(srcDir.toString(), new Path(dtDir.toString()), destDir + "/" + dirName);
         }
       }
     }
     return destFiles;
   }

   static Collection<Path> processDatasourceDirectory(String clusterpath, Path srcDir,
       String destDir) throws Exception {
     Path cPath = new Path(clusterpath);
     String cluster = cPath.getName();

     Collection<Path> destFiles = new HashSet<Path>();
     String fileName = null;
     int fileDay = 0;
     int fileHour = 0;
     int fileMin = 0;

     FileStatus[] recordFiles = fs.listStatus(srcDir);
     for (FileStatus recordFile : recordFiles) {
       // dataSource_20080915_18_15.1.evt
       // <datasource>_<yyyyMMdd_HH_mm>.1.evt

       fileName = recordFile.getPath().getName();
       log.info("processDatasourceDirectory processing RecordFile: (" + fileName
           + ")");
       log.info("fileName: " + fileName);

       int l = fileName.length();
       String dataSource = HierarchyDataType.getDataType(srcDir, cPath);
       log.info("Datasource: " + dataSource);

       if (fileName.endsWith(".D.evt")) {
         // Hadoop_dfs_datanode_20080919.D.evt

         fileDay = Integer.parseInt(fileName.substring(l - 14, l - 6));
         Path destFile = writeRecordFile(destDir + "/" + fileDay + "/",
             recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
                 + fileDay);
         if (destFile != null) {
           destFiles.add(destFile);
         }
       } else if (fileName.endsWith(".H.evt")) {
         // Hadoop_dfs_datanode_20080925_1.H.evt
         // Hadoop_dfs_datanode_20080925_12.H.evt

         String day = null;
         String hour = null;
         if (fileName.charAt(l - 8) == '_') {
           day = fileName.substring(l - 16, l - 8);
           log.info("day->" + day);
           hour = "" + fileName.charAt(l - 7);
           log.info("hour->" + hour);
         } else {
           day = fileName.substring(l - 17, l - 9);
           log.info("day->" + day);
           hour = fileName.substring(l - 8, l - 6);
           log.info("hour->" + hour);
         }
         fileDay = Integer.parseInt(day);
         fileHour = Integer.parseInt(hour);
         // rotate there so spill
         Path destFile = writeRecordFile(destDir + "/" + fileDay + "/"
             + fileHour + "/", recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
                 + fileDay + "_" + fileHour);
         if (destFile != null) {
           destFiles.add(destFile);
         }
         // mark this directory for daily rotate
         addDirectory4Rolling(true, fileDay, fileHour, cluster, dataSource);
       } else if (fileName.endsWith(".R.evt")) {
         if (fileName.charAt(l - 11) == '_') {
           fileDay = Integer.parseInt(fileName.substring(l - 19, l - 11));
           fileHour = Integer.parseInt("" + fileName.charAt(l - 10));
           fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
         } else {
           fileDay = Integer.parseInt(fileName.substring(l - 20, l - 12));
           fileHour = Integer.parseInt(fileName.substring(l - 11, l - 9));
           fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
         }

         log.info("fileDay: " + fileDay);
         log.info("fileHour: " + fileHour);
         log.info("fileMin: " + fileMin);
         Path destFile = writeRecordFile(
             destDir + "/" + fileDay + "/" + fileHour + "/" + fileMin,
             recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(HierarchyDataType.trimSlash(dataSource))
             + "_" + fileDay + "_" + fileHour + "_" + fileMin);
         if (destFile != null) {
           destFiles.add(destFile);
         }
         // mark this directory for hourly rotate
         addDirectory4Rolling(false, fileDay, fileHour, cluster, dataSource);
       } else {
         throw new RuntimeException("Wrong fileName format! [" + fileName + "]");
       }
     }

     return destFiles;
   }

   static void addDirectory4Rolling(boolean isDailyOnly, int day, int hour,
       String cluster, String dataSource) throws IOException {
     // TODO get root directory from config
     String rollingDirectory = "/chukwa/rolling/";

     Path path = new Path(rollingDirectory + "/daily/" + day + "/" + cluster
         + "/" + dataSource);
     if (!fs.exists(path)) {
       fs.mkdirs(path);
     }

     if (!isDailyOnly) {
       path = new Path(rollingDirectory + "/hourly/" + day + "/" + hour + "/"
           + cluster + "/" + dataSource);
       if (!fs.exists(path)) {
         fs.mkdirs(path);
       }
     }
   }

   static Path writeRecordFile(String destDir, Path recordFile, String fileName)
       throws IOException {
     boolean done = false;
     int count = 1;
     do {
       Path destDirPath = new Path(destDir);
       Path destFilePath = new Path(destDir + "/" + fileName + "." + count
           + ".evt");

       if (!fs.exists(destDirPath)) {
         fs.mkdirs(destDirPath);
         log.info(">>>>>>>>>>>> create Dir" + destDirPath);
       }

       if (!fs.exists(destFilePath)) {
         log.info(">>>>>>>>>>>> Before Rename" + recordFile + " -- "
             + destFilePath);
         boolean rename = fs.rename(recordFile,destFilePath);
         done = true;
         log.info(">>>>>>>>>>>> after Rename" + destFilePath + " , rename:"+rename);
         return destFilePath;
       }
       count++;

       if (count > 1000) {
         log.warn("too many files in this directory: " + destDir);
       }
     } while (!done);

     return null;
   }

   static boolean checkRotate(String directoryAsString,
       boolean createDirectoryIfNotExist) throws IOException {
     Path directory = new Path(directoryAsString);
     boolean exist = fs.exists(directory);

     if (!exist) {
       if (createDirectoryIfNotExist == true) {
         fs.mkdirs(directory);
       }
       return false;
     } else {
       return fs.exists(new Path(directoryAsString + "/rotateDone"));
     }
   }

   public static Path[] doMove(Path srcDir, String destDir) throws Exception {
     conf = new ChukwaConfiguration();
     String fsName = conf.get("writer.hdfs.filesystem");
     fs = FileSystem.get(new URI(fsName), conf);
     log.info("Start MoveToRepository doMove()");

     FileStatus fstat = fs.getFileStatus(srcDir);

     Collection<Path> destinationFiles = new HashSet<Path>();
     if (!fstat.isDir()) {
       throw new IOException(srcDir + " is not a directory!");
     } else {
       FileStatus[] clusters = fs.listStatus(srcDir);
       // Run a moveOrMerge on all clusters
       String name = null;
       for (FileStatus cluster : clusters) {
         name = cluster.getPath().getName();
         // Skip hadoop M/R outputDir
         if (name.startsWith("_")) {
           continue;
         }
         log.info("main procesing Cluster (" + cluster.getPath().getName() + ")");
         destinationFiles.addAll(processClusterDirectory(cluster.getPath(),
             destDir + "/" + cluster.getPath().getName()));

         // Delete the demux's cluster dir
         FileUtil.fullyDelete(fs, cluster.getPath());
       }
     }

     log.info("Done with MoveToRepository doMove()");
     return destinationFiles.toArray(new Path[destinationFiles.size()]);
   }

   /**
    * @param args is command line parameter
    * @throws Exception if error in processing data
    */
   public static void main(String[] args) throws Exception {

     Path srcDir = new Path(args[0]);
     String destDir = args[1];
     doMove(srcDir, destDir);
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.chukwa.extraction.demux;


	import java.io.IOException;
	import java.net.URI;
	import java.text.SimpleDateFormat;
	import java.util.Calendar;
	import java.util.Collection;
	import java.util.HashSet;
	import java.util.List;

	import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
	import org.apache.hadoop.chukwa.util.HierarchyDataType;
	import org.apache.hadoop.fs.PathFilter;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.FileUtil;
	import org.apache.hadoop.fs.Path;
	import org.apache.log4j.Logger;

	// TODO
	// First version of the Spill
	// need some polishing

	public class MoveToRepository {
	static Logger log = Logger.getLogger(MoveToRepository.class);

	static ChukwaConfiguration conf = null;
	static FileSystem fs = null;
	static SimpleDateFormat sdf = new java.text.SimpleDateFormat("yyyyMMdd");
	static Calendar calendar = Calendar.getInstance();

	static Collection<Path> processClusterDirectory(Path srcDir, String destDir)
	throws Exception {
	log.info("processClusterDirectory (" + srcDir.getName() + "," + destDir
	+ ")");
	FileStatus fstat = fs.getFileStatus(srcDir);
	Collection<Path> destFiles = new HashSet<Path>();

	if (!fstat.isDir()) {
	throw new IOException(srcDir + " is not a directory!");
	} else {
	FileStatus[] datasourceDirectories = fs.listStatus(srcDir);

	for (FileStatus datasourceDirectory : datasourceDirectories) {
	log.info(datasourceDirectory.getPath() + " isDir?"
	+ datasourceDirectory.isDir());
	if (!datasourceDirectory.isDir()) {
	throw new IOException(
	"Top level datasource directory should be a directory :"
	+ datasourceDirectory.getPath());
	}

	PathFilter filter = new PathFilter()
	{public boolean accept(Path file) {
	return file.getName().endsWith(".evt");
	} };
	//CHUKWA-648: Make Chukwa Reduce Type to support hierarchy format
	// to processDataSourceDirectory according to hierarchy data type format
	List<FileStatus> eventfiles = HierarchyDataType.globStatus(fs, datasourceDirectory.getPath(),filter,true);
	for (FileStatus eventfile : eventfiles){
	Path datatypeDir = eventfile.getPath().getParent();
	String dirName = HierarchyDataType.getDataType(datatypeDir, srcDir);

	Path destPath = new Path(destDir + "/" + dirName);
	log.info("dest directory path: " + destPath);
	log.info("processClusterDirectory processing Datasource: (" + dirName
	+ ")");
	StringBuilder dtDir = new StringBuilder(srcDir.toString()).append("/").append(dirName);
	log.debug("srcDir: " + dtDir.toString());
	processDatasourceDirectory(srcDir.toString(), new Path(dtDir.toString()), destDir + "/" + dirName);
	}
	}
	}
	return destFiles;
	}

	static Collection<Path> processDatasourceDirectory(String clusterpath, Path srcDir,
	String destDir) throws Exception {
	Path cPath = new Path(clusterpath);
	String cluster = cPath.getName();

	Collection<Path> destFiles = new HashSet<Path>();
	String fileName = null;
	int fileDay = 0;
	int fileHour = 0;
	int fileMin = 0;

	FileStatus[] recordFiles = fs.listStatus(srcDir);
	for (FileStatus recordFile : recordFiles) {
	// dataSource_20080915_18_15.1.evt
	// <datasource>_<yyyyMMdd_HH_mm>.1.evt

	fileName = recordFile.getPath().getName();
	log.info("processDatasourceDirectory processing RecordFile: (" + fileName
	+ ")");
	log.info("fileName: " + fileName);

	int l = fileName.length();
	String dataSource = HierarchyDataType.getDataType(srcDir, cPath);
	log.info("Datasource: " + dataSource);

	if (fileName.endsWith(".D.evt")) {
	// Hadoop_dfs_datanode_20080919.D.evt

	fileDay = Integer.parseInt(fileName.substring(l - 14, l - 6));
	Path destFile = writeRecordFile(destDir + "/" + fileDay + "/",
	recordFile.getPath(),
	HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
	+ fileDay);
	if (destFile != null) {
	destFiles.add(destFile);
	}
	} else if (fileName.endsWith(".H.evt")) {
	// Hadoop_dfs_datanode_20080925_1.H.evt
	// Hadoop_dfs_datanode_20080925_12.H.evt

	String day = null;
	String hour = null;
	if (fileName.charAt(l - 8) == '_') {
	day = fileName.substring(l - 16, l - 8);
	log.info("day->" + day);
	hour = "" + fileName.charAt(l - 7);
	log.info("hour->" + hour);
	} else {
	day = fileName.substring(l - 17, l - 9);
	log.info("day->" + day);
	hour = fileName.substring(l - 8, l - 6);
	log.info("hour->" + hour);
	}
	fileDay = Integer.parseInt(day);
	fileHour = Integer.parseInt(hour);
	// rotate there so spill
	Path destFile = writeRecordFile(destDir + "/" + fileDay + "/"
	+ fileHour + "/", recordFile.getPath(),
	HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
	+ fileDay + "_" + fileHour);
	if (destFile != null) {
	destFiles.add(destFile);
	}
	// mark this directory for daily rotate
	addDirectory4Rolling(true, fileDay, fileHour, cluster, dataSource);
	} else if (fileName.endsWith(".R.evt")) {
	if (fileName.charAt(l - 11) == '_') {
	fileDay = Integer.parseInt(fileName.substring(l - 19, l - 11));
	fileHour = Integer.parseInt("" + fileName.charAt(l - 10));
	fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
	} else {
	fileDay = Integer.parseInt(fileName.substring(l - 20, l - 12));
	fileHour = Integer.parseInt(fileName.substring(l - 11, l - 9));
	fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
	}

	log.info("fileDay: " + fileDay);
	log.info("fileHour: " + fileHour);
	log.info("fileMin: " + fileMin);
	Path destFile = writeRecordFile(
	destDir + "/" + fileDay + "/" + fileHour + "/" + fileMin,
	recordFile.getPath(),
	HierarchyDataType.getHierarchyDataTypeFileName(HierarchyDataType.trimSlash(dataSource))
	+ "_" + fileDay + "_" + fileHour + "_" + fileMin);
	if (destFile != null) {
	destFiles.add(destFile);
	}
	// mark this directory for hourly rotate
	addDirectory4Rolling(false, fileDay, fileHour, cluster, dataSource);
	} else {
	throw new RuntimeException("Wrong fileName format! [" + fileName + "]");
	}
	}

	return destFiles;
	}

	static void addDirectory4Rolling(boolean isDailyOnly, int day, int hour,
	String cluster, String dataSource) throws IOException {
	// TODO get root directory from config
	String rollingDirectory = "/chukwa/rolling/";

	Path path = new Path(rollingDirectory + "/daily/" + day + "/" + cluster
	+ "/" + dataSource);
	if (!fs.exists(path)) {
	fs.mkdirs(path);
	}

	if (!isDailyOnly) {
	path = new Path(rollingDirectory + "/hourly/" + day + "/" + hour + "/"
	+ cluster + "/" + dataSource);
	if (!fs.exists(path)) {
	fs.mkdirs(path);
	}
	}
	}

	static Path writeRecordFile(String destDir, Path recordFile, String fileName)
	throws IOException {
	boolean done = false;
	int count = 1;
	do {
	Path destDirPath = new Path(destDir);
	Path destFilePath = new Path(destDir + "/" + fileName + "." + count
	+ ".evt");

	if (!fs.exists(destDirPath)) {
	fs.mkdirs(destDirPath);
	log.info(">>>>>>>>>>>> create Dir" + destDirPath);
	}

	if (!fs.exists(destFilePath)) {
	log.info(">>>>>>>>>>>> Before Rename" + recordFile + " -- "
	+ destFilePath);
	boolean rename = fs.rename(recordFile,destFilePath);
	done = true;
	log.info(">>>>>>>>>>>> after Rename" + destFilePath + " , rename:"+rename);
	return destFilePath;
	}
	count++;

	if (count > 1000) {
	log.warn("too many files in this directory: " + destDir);
	}
	} while (!done);

	return null;
	}

	static boolean checkRotate(String directoryAsString,
	boolean createDirectoryIfNotExist) throws IOException {
	Path directory = new Path(directoryAsString);
	boolean exist = fs.exists(directory);

	if (!exist) {
	if (createDirectoryIfNotExist == true) {
	fs.mkdirs(directory);
	}
	return false;
	} else {
	return fs.exists(new Path(directoryAsString + "/rotateDone"));
	}
	}

	public static Path[] doMove(Path srcDir, String destDir) throws Exception {
	conf = new ChukwaConfiguration();
	String fsName = conf.get("writer.hdfs.filesystem");
	fs = FileSystem.get(new URI(fsName), conf);
	log.info("Start MoveToRepository doMove()");

	FileStatus fstat = fs.getFileStatus(srcDir);

	Collection<Path> destinationFiles = new HashSet<Path>();
	if (!fstat.isDir()) {
	throw new IOException(srcDir + " is not a directory!");
	} else {
	FileStatus[] clusters = fs.listStatus(srcDir);
	// Run a moveOrMerge on all clusters
	String name = null;
	for (FileStatus cluster : clusters) {
	name = cluster.getPath().getName();
	// Skip hadoop M/R outputDir
	if (name.startsWith("_")) {
	continue;
	}
	log.info("main procesing Cluster (" + cluster.getPath().getName() + ")");
	destinationFiles.addAll(processClusterDirectory(cluster.getPath(),
	destDir + "/" + cluster.getPath().getName()));

	// Delete the demux's cluster dir
	FileUtil.fullyDelete(fs, cluster.getPath());
	}
	}

	log.info("Done with MoveToRepository doMove()");
	return destinationFiles.toArray(new Path[destinationFiles.size()]);
	}

	/**
	* @param args is command line parameter
	* @throws Exception if error in processing data
	*/
	public static void main(String[] args) throws Exception {

	Path srcDir = new Path(args[0]);
	String destDir = args[1];
	doMove(srcDir, destDir);
	}

	}