public static void cleanJobStagingData ( State state , Logger logger ) throws IOException { Preconditions . checkArgument ( state . contains ( ConfigurationKeys . WRITER_STAGING_DIR ) , "Missing required property " + ConfigurationKeys . WRITER_STAGING_DIR ) ; Preconditions . checkArgument ( state . contains ( ConfigurationKeys . WRITER_OUTPUT_DIR ) , "Missing required property " + ConfigurationKeys . WRITER_OUTPUT_DIR ) ; String writerFsUri = state . getProp ( ConfigurationKeys . WRITER_FILE_SYSTEM_URI , ConfigurationKeys . LOCAL_FS_URI ) ; FileSystem fs = getFsWithProxy ( state , writerFsUri , WriterUtils . getFsConfiguration ( state ) ) ; Path jobStagingPath = new Path ( state . getProp ( ConfigurationKeys . WRITER_STAGING_DIR ) ) ; logger . info ( "Cleaning up staging directory " + jobStagingPath ) ; HadoopUtils . deletePath ( fs , jobStagingPath , true ) ; if ( fs . exists ( jobStagingPath . getParent ( ) ) && fs . listStatus ( jobStagingPath . getParent ( ) ) . length == 0 ) { logger . info ( "Deleting directory " + jobStagingPath . getParent ( ) ) ; HadoopUtils . deletePath ( fs , jobStagingPath . getParent ( ) , true ) ; } Path jobOutputPath = new Path ( state . getProp ( ConfigurationKeys . WRITER_OUTPUT_DIR ) ) ; logger . info ( "Cleaning up output directory " + jobOutputPath ) ; HadoopUtils . deletePath ( fs , jobOutputPath , true ) ; if ( fs . exists ( jobOutputPath . getParent ( ) ) && fs . listStatus ( jobOutputPath . getParent ( ) ) . length == 0 ) { logger . info ( "Deleting directory " + jobOutputPath . getParent ( ) ) ; HadoopUtils . deletePath ( fs , jobOutputPath . getParent ( ) , true ) ; } if ( state . contains ( ConfigurationKeys . ROW_LEVEL_ERR_FILE ) ) { if ( state . getPropAsBoolean ( ConfigurationKeys . CLEAN_ERR_DIR , ConfigurationKeys . DEFAULT_CLEAN_ERR_DIR ) ) { Path jobErrPath = new Path ( state . getProp ( ConfigurationKeys . ROW_LEVEL_ERR_FILE ) ) ; log . info ( "Cleaning up err directory : " + jobErrPath ) ; HadoopUtils . deleteIfExists ( fs , jobErrPath , true ) ; } } }
Cleanup staging data of all tasks of a job .