public BlockOutStream getOutStream ( long blockId , long blockSize , OutStreamOptions options ) throws IOException { WorkerNetAddress address ; BlockLocationPolicy locationPolicy = Preconditions . checkNotNull ( options . getLocationPolicy ( ) , PreconditionMessage . BLOCK_WRITE_LOCATION_POLICY_UNSPECIFIED ) ; GetWorkerOptions workerOptions = GetWorkerOptions . defaults ( ) . setBlockInfo ( new BlockInfo ( ) . setBlockId ( blockId ) . setLength ( blockSize ) ) . setBlockWorkerInfos ( new ArrayList <> ( getEligibleWorkers ( ) ) ) ; // The number of initial copies depends on the write type: if ASYNC_THROUGH, it is the property // "alluxio.user.file.replication.durable" before data has been persisted; otherwise // "alluxio.user.file.replication.min" int initialReplicas = ( options . getWriteType ( ) == WriteType . ASYNC_THROUGH && options . getReplicationDurable ( ) > options . getReplicationMin ( ) ) ? options . getReplicationDurable ( ) : options . getReplicationMin ( ) ; if ( initialReplicas <= 1 ) { address = locationPolicy . getWorker ( workerOptions ) ; if ( address == null ) { throw new UnavailableException ( ExceptionMessage . NO_SPACE_FOR_BLOCK_ON_WORKER . getMessage ( blockSize ) ) ; } return getOutStream ( blockId , blockSize , address , options ) ; } // Group different block workers by their hostnames Map < String , Set < BlockWorkerInfo > > blockWorkersByHost = new HashMap <> ( ) ; for ( BlockWorkerInfo blockWorker : workerOptions . getBlockWorkerInfos ( ) ) { String hostName = blockWorker . getNetAddress ( ) . getHost ( ) ; if ( blockWorkersByHost . containsKey ( hostName ) ) { blockWorkersByHost . get ( hostName ) . add ( blockWorker ) ; } else { blockWorkersByHost . put ( hostName , com . google . common . collect . Sets . newHashSet ( blockWorker ) ) ; } } // Select N workers on different hosts where N is the value of initialReplicas for this block List < WorkerNetAddress > workerAddressList = new ArrayList <> ( ) ; List < BlockWorkerInfo > updatedInfos = Lists . newArrayList ( workerOptions . getBlockWorkerInfos ( ) ) ; for ( int i = 0 ; i < initialReplicas ; i ++ ) { address = locationPolicy . getWorker ( workerOptions ) ; if ( address == null ) { break ; } workerAddressList . add ( address ) ; updatedInfos . removeAll ( blockWorkersByHost . get ( address . getHost ( ) ) ) ; workerOptions . setBlockWorkerInfos ( updatedInfos ) ; } if ( workerAddressList . size ( ) < initialReplicas ) { throw new alluxio . exception . status . ResourceExhaustedException ( String . format ( "Not enough workers for replications, %d workers selected but %d required" , workerAddressList . size ( ) , initialReplicas ) ) ; } return BlockOutStream . createReplicatedBlockOutStream ( mContext , blockId , blockSize , workerAddressList , options ) ; }
Gets a stream to write data to a block based on the options . The stream can only be backed by Alluxio storage .