private Path copyRemoteFiles ( FileSystem jtFs , Path parentDir , Path originalPath , JobConf job , short replication , String md5 ) throws IOException { //check if we do not need to copy the files // is jt using the same file system. // just checking for uri strings... doing no dns lookups // to see if the filesystems are the same. This is not optimal. // but avoids name resolution. FileSystem remoteFs = null ; remoteFs = originalPath . getFileSystem ( job ) ; if ( compareFs ( remoteFs , jtFs ) ) { return originalPath ; } // This function is overloaded to support cache sharing when enabled if ( md5 != null ) { // Check if file already exists in cache Path basePath = parentDir ; Path realPath = new Path ( basePath , md5 + "_" + originalPath . getName ( ) ) ; Path qualifiedRealPath = realPath . makeQualified ( jtFs ) ; if ( filesInCache . contains ( qualifiedRealPath ) ) { // We "touch" the file to update its access time // This is done only 10% of the time to reduce load on the namenode if ( r . nextLong ( ) % 10 == 0 ) { try { jtFs . setTimes ( realPath , - 1 , System . currentTimeMillis ( ) ) ; } catch ( RemoteException e ) { LOG . warn ( "Error in setTimes" , e ) ; } } return qualifiedRealPath ; } // This loop should not even loop most of the time Path newPath ; do { newPath = new Path ( basePath , "tmp_" + originalPath . getName ( ) + r . nextLong ( ) ) ; } while ( jtFs . exists ( newPath ) ) ; FileUtil . copy ( remoteFs , originalPath , jtFs , newPath , false , job ) ; jtFs . setReplication ( newPath , replication ) ; jtFs . setPermission ( newPath , new FsPermission ( JOB_DIR_PERMISSION ) ) ; LOG . info ( "Uploading new shared jar: " + realPath . toString ( ) ) ; if ( ! jtFs . rename ( newPath , realPath ) ) { // if there are multiple clients racing to upload the new jar - only // one of them will succeed. Check if we failed because the file already // exists. if so, ignore and move on if ( ! jtFs . exists ( realPath ) ) throw new IOException ( "Unable to upload or find shared jar: " + realPath . toString ( ) ) ; } // Update the list filesInCache . add ( qualifiedRealPath ) ; return qualifiedRealPath ; } // this might have name collisions. copy will throw an exception // parse the original path to create new path Path newPath = new Path ( parentDir , originalPath . getName ( ) ) ; FileUtil . copy ( remoteFs , originalPath , jtFs , newPath , false , job ) ; jtFs . setReplication ( newPath , replication ) ; return jtFs . makeQualified ( newPath ) ; }
was copied to