public static void loadBlock ( FileSystem fs , FileSystemContext context , String path , long blockId ) throws AlluxioException , IOException { AlluxioBlockStore blockStore = AlluxioBlockStore . create ( context ) ; String localHostName = NetworkAddressUtils . getConnectHost ( ServiceType . WORKER_RPC , ServerConfiguration . global ( ) ) ; List < BlockWorkerInfo > workerInfoList = blockStore . getAllWorkers ( ) ; WorkerNetAddress localNetAddress = null ; for ( BlockWorkerInfo workerInfo : workerInfoList ) { if ( workerInfo . getNetAddress ( ) . getHost ( ) . equals ( localHostName ) ) { localNetAddress = workerInfo . getNetAddress ( ) ; break ; } } if ( localNetAddress == null ) { throw new NotFoundException ( ExceptionMessage . NO_LOCAL_BLOCK_WORKER_REPLICATE_TASK . getMessage ( blockId ) ) ; } // TODO(jiri): Replace with internal client that uses file ID once the internal client is // factored out of the core server module. The reason to prefer using file ID for this job is // to avoid the the race between "replicate" and "rename", so that even a file to replicate is // renamed, the job is still working on the correct file. URIStatus status = fs . getStatus ( new AlluxioURI ( path ) ) ; OpenFilePOptions openOptions = OpenFilePOptions . newBuilder ( ) . setReadType ( ReadPType . NO_CACHE ) . build ( ) ; AlluxioConfiguration conf = ServerConfiguration . global ( ) ; InStreamOptions inOptions = new InStreamOptions ( status , openOptions , conf ) ; // Set read location policy always to loca first for loading blocks for job tasks inOptions . setUfsReadLocationPolicy ( BlockLocationPolicy . Factory . create ( LocalFirstPolicy . class . getCanonicalName ( ) , conf ) ) ; OutStreamOptions outOptions = OutStreamOptions . defaults ( conf ) ; // Set write location policy always to local first for loading blocks for job tasks outOptions . setLocationPolicy ( BlockLocationPolicy . Factory . create ( LocalFirstPolicy . class . getCanonicalName ( ) , conf ) ) ; // use -1 to reuse the existing block size for this block try ( OutputStream outputStream = blockStore . getOutStream ( blockId , - 1 , localNetAddress , outOptions ) ) { try ( InputStream inputStream = blockStore . getInStream ( blockId , inOptions ) ) { ByteStreams . copy ( inputStream , outputStream ) ; } catch ( Throwable t ) { try { ( ( Cancelable ) outputStream ) . cancel ( ) ; } catch ( Throwable t2 ) { t . addSuppressed ( t2 ) ; } throw t ; } } }
Loads a block into the local worker . If the block doesn t exist in Alluxio it will be read from the UFS .