private int distRaid(String[] args, int startIndex) throws IOException,
    SAXException, RaidConfigurationException,
    ClassNotFoundException, ParserConfigurationException, JSONException {
    // find the matched raid policy
    String policyName = args[startIndex++];
    ConfigManager configManager = new ConfigManager(conf);
    PolicyInfo policy = configManager.getPolicy(policyName);
    if (policy == null) {
      System.err.println ("Invalid policy: " + policyName);
      return -1;
    }
    Codec codec = Codec.getCodec(policy.getCodecId());
    if (codec == null) {
      System.err.println("Policy " + policyName
          + " with invalid codec " + policy.getCodecId());
    }

    // find the matched paths to raid
    FileSystem fs = FileSystem.get(conf);
    List<FileStatus> pathsToRaid = new ArrayList<FileStatus>();
    List<Path> policySrcPaths = policy.getSrcPathExpanded();
    for (int i = startIndex; i< args.length; i++) {
      boolean invalidPathToRaid = true;
      Path pathToRaid = new Path(args[i]).makeQualified(fs);
      String pathToRaidStr = pathToRaid.toString();
      if (!pathToRaidStr.endsWith(Path.SEPARATOR)) {
        pathToRaidStr = pathToRaidStr.concat(Path.SEPARATOR);
      }
      for (Path srcPath : policySrcPaths) {
        String srcStr = srcPath.toString();
        if (!srcStr.endsWith(Path.SEPARATOR)) {
          srcStr = srcStr.concat(Path.SEPARATOR);
        }
        if (pathToRaidStr.startsWith(srcStr)) {
          if (codec.isDirRaid) {
            FileUtil.listStatusForLeafDir(
                fs, fs.getFileStatus(pathToRaid), pathsToRaid);
          } else {
            FileUtil.listStatusHelper(fs, pathToRaid, 
                Integer.MAX_VALUE, pathsToRaid);
          }
          invalidPathToRaid = false;
          break;
        }
      }
      if (invalidPathToRaid) {
        System.err.println("Path " + pathToRaidStr + 
          " does not support by the given policy " + policyName);
      }
    }
    
    // Check if files are valid
    List<FileStatus> validPaths = new ArrayList<FileStatus>();
    List<PolicyInfo> policyInfos = new ArrayList<PolicyInfo>(1);
    policyInfos.add(policy);
    RaidState.Checker checker = new RaidState.Checker(
        policyInfos, conf);
    long now = System.currentTimeMillis();
    for (FileStatus fileStatus : pathsToRaid) {
      FileStatus[] dirStats = null;
      if (codec.isDirRaid) {
        dirStats = fs.listStatus(fileStatus.getPath());
      }
      RaidState stat = checker.check(
          policy, fileStatus, now, false, 
          dirStats == null ? null : Arrays.asList(dirStats));
      if (stat == RaidState.NOT_RAIDED_BUT_SHOULD) {
        validPaths.add(fileStatus);
      } else {
        System.err.println("Path " + fileStatus.getPath() + 
            " is not qualified for raiding: " + stat);
      }
    }
    if (validPaths.isEmpty()) {
      System.err.println("No file can be raided");
      return 0;
    }
    DistRaid dr = new DistRaid(conf);
    //add paths for distributed raiding
    List<EncodingCandidate> validEC =
        RaidNode.splitPaths(conf, Codec.getCodec(policy.getCodecId()), validPaths);
    dr.addRaidPaths(policy, validEC);
    
    if (dr.startDistRaid()) {
      System.out.println("Job started: " + dr.getJobTrackingURL());
      System.out.print("Job in progress ");
      while (!dr.checkComplete()) {
        try {
          System.out.print(".");
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          throw new InterruptedIOException("Got interrupted.");
        }
      }
      if (dr.successful()) {
        System.out.println("/nFiles are successfully raided.");
        return 0;
      } else {
        System.err.println("/nRaid job failed.");
        return -1;
      }
    }
    return -1;
  }
private int distRaid ( String [ ] args , int startIndex ) throws IOException , SAXException , RaidConfigurationException , ClassNotFoundException , ParserConfigurationException , JSONException { String policyName = args [ startIndex ++ ] ; ConfigManager configManager = new ConfigManager ( conf ) ; PolicyInfo policy = configManager . getPolicy ( policyName ) ; if ( policy == null ) { System . err . println ( "Invalid policy: " + policyName ) ; return - 1 ; } Codec codec = Codec . getCodec ( policy . getCodecId ( ) ) ; if ( codec == null ) { System . err . println ( "Policy " + policyName + " with invalid codec " + policy . getCodecId ( ) ) ; } FileSystem fs = FileSystem . get ( conf ) ; List < FileStatus > pathsToRaid = new ArrayList < FileStatus > ( ) ; List < Path > policySrcPaths = policy . getSrcPathExpanded ( ) ; for ( int i = startIndex ; i < args . length ; i ++ ) { boolean invalidPathToRaid = true ; Path pathToRaid = new Path ( args [ i ] ) . makeQualified ( fs ) ; String pathToRaidStr = pathToRaid . toString ( ) ; if ( ! pathToRaidStr . endsWith ( Path . SEPARATOR ) ) { pathToRaidStr = pathToRaidStr . concat ( Path . SEPARATOR ) ; } for ( Path srcPath : policySrcPaths ) { String srcStr = srcPath . toString ( ) ; if ( ! srcStr . endsWith ( Path . SEPARATOR ) ) { srcStr = srcStr . concat ( Path . SEPARATOR ) ; } if ( pathToRaidStr . startsWith ( srcStr ) ) { if ( codec . isDirRaid ) { FileUtil . listStatusForLeafDir ( fs , fs . getFileStatus ( pathToRaid ) , pathsToRaid ) ; } else { FileUtil . listStatusHelper ( fs , pathToRaid , Integer . MAX_VALUE , pathsToRaid ) ; } invalidPathToRaid = false ; break ; } } if ( invalidPathToRaid ) { System . err . println ( "Path " + pathToRaidStr + " does not support by the given policy " + policyName ) ; } } List < FileStatus > validPaths = new ArrayList < FileStatus > ( ) ; List < PolicyInfo > policyInfos = new ArrayList < PolicyInfo > ( 1 ) ; policyInfos . add ( policy ) ; RaidState . Checker checker = new RaidState . Checker ( policyInfos , conf ) ; long now = System . currentTimeMillis ( ) ; for ( FileStatus fileStatus : pathsToRaid ) { FileStatus [ ] dirStats = null ; if ( codec . isDirRaid ) { dirStats = fs . listStatus ( fileStatus . getPath ( ) ) ; } RaidState stat = checker . check ( policy , fileStatus , now , false , dirStats == null ? null : Arrays . asList ( dirStats ) ) ; if ( stat == RaidState . NOT_RAIDED_BUT_SHOULD ) { validPaths . add ( fileStatus ) ; } else { System . err . println ( "Path " + fileStatus . getPath ( ) + " is not qualified for raiding: " + stat ) ; } } if ( validPaths . isEmpty ( ) ) { System . err . println ( "No file can be raided" ) ; return 0 ; } DistRaid dr = new DistRaid ( conf ) ; List < EncodingCandidate > validEC = RaidNode . splitPaths ( conf , Codec . getCodec ( policy . getCodecId ( ) ) , validPaths ) ; dr . addRaidPaths ( policy , validEC ) ; if ( dr . startDistRaid ( ) ) { System . out . println ( "Job started: " + dr . getJobTrackingURL ( ) ) ; System . out . print ( "Job in progress " ) ; while ( ! dr . checkComplete ( ) ) { try { System . out . print ( "." ) ; Thread . sleep ( 1000 ) ; } catch ( InterruptedException e ) { throw new InterruptedIOException ( "Got interrupted." ) ; } } if ( dr . successful ( ) ) { System . out . println ( "/nFiles are successfully raided." ) ; return 0 ; } else { System . err . println ( "/nRaid job failed." ) ; return - 1 ; } } return - 1 ; }
Submit a map/reduce job to raid the input paths
@param args all input parameters
@param startIndex staring index of arguments: policy_name path1, ..., pathn
@return 0 if successful
@throws IOException if any error occurs
@throws ParserConfigurationException
@throws ClassNotFoundException
@throws RaidConfigurationException
@throws SAXException