I just boot a demon thread scheduleAtFixRate clean these "empty" and has file "_SUCCESS" directory and another thread to run hive cmd "alter xxx concatenate" Executors . newSingleThreadScheduledExecutor ().scheduleAtFixedRate( new Runnable { override def run(): Unit = { val fs = FileSystem. get ( new Configuration()) val status = fs.listStatus( new Path( s"hdfs://nameservice/user/xxx/warehouse/ $ tableName /" )) status.foreach(stat => if (stat.isDirectory && stat.getPath.getName.contains( "hive-staging" ) && fs.getContentSummary (stat.getPath).getSpaceConsumed < 1024 ) { println ( "empty path : " + stat.getPath) if ( directoryHasSuccess (stat.getPath, fs)) { fs.delete(stat.getPath, true ) } val now = new Date().getTime if (now - stat.getModificationTime > 5 * 60 * 1000 && (now - stat.getAccessTime > 5 * 60 * 1000 )) { //5m before println ( "delete path " + stat.getPath) fs.delete(stat.getPath, true ) } } ) } }, 5 , interval , TimeUnit. SECONDS );
... View more
