@ Override protected void serviceInit ( Configuration conf ) throws Exception { _conf = conf ; boolean stopOnFailure = conf . getBoolean ( STOP_ON_FAILURE_KEY , DEFAULT_STOP_ON_FAILURE ) ; try { // In case this NM was killed while there were running spark applications, we need to restore // lost state for the existing executors. We look for an existing file in the NM's local dirs. // If we don't find one, then we choose a file to use to save the state next time. Even if // an application was stopped while the NM was down, we expect yarn to call stopApplication() // when it comes back if ( _recoveryPath != null ) { registeredExecutorFile = initRecoveryDb ( RECOVERY_FILE_NAME ) ; } TransportConf transportConf = new TransportConf ( "shuffle" , new HadoopConfigProvider ( conf ) ) ; blockHandler = new ExternalShuffleBlockHandler ( transportConf , registeredExecutorFile ) ; // If authentication is enabled, set up the shuffle server to use a // special RPC handler that filters out unauthenticated fetch requests List < TransportServerBootstrap > bootstraps = Lists . newArrayList ( ) ; boolean authEnabled = conf . getBoolean ( SPARK_AUTHENTICATE_KEY , DEFAULT_SPARK_AUTHENTICATE ) ; if ( authEnabled ) { secretManager = new ShuffleSecretManager ( ) ; if ( _recoveryPath != null ) { loadSecretsFromDb ( ) ; } bootstraps . add ( new AuthServerBootstrap ( transportConf , secretManager ) ) ; } int port = conf . getInt ( SPARK_SHUFFLE_SERVICE_PORT_KEY , DEFAULT_SPARK_SHUFFLE_SERVICE_PORT ) ; transportContext = new TransportContext ( transportConf , blockHandler ) ; shuffleServer = transportContext . createServer ( port , bootstraps ) ; // the port should normally be fixed, but for tests its useful to find an open port port = shuffleServer . getPort ( ) ; boundPort = port ; String authEnabledString = authEnabled ? "enabled" : "not enabled" ; // register metrics on the block handler into the Node Manager's metrics system. blockHandler . getAllMetrics ( ) . getMetrics ( ) . put ( "numRegisteredConnections" , shuffleServer . getRegisteredConnections ( ) ) ; YarnShuffleServiceMetrics serviceMetrics = new YarnShuffleServiceMetrics ( blockHandler . getAllMetrics ( ) ) ; MetricsSystemImpl metricsSystem = ( MetricsSystemImpl ) DefaultMetricsSystem . instance ( ) ; metricsSystem . register ( "sparkShuffleService" , "Metrics on the Spark Shuffle Service" , serviceMetrics ) ; logger . info ( "Registered metrics with Hadoop's DefaultMetricsSystem" ) ; logger . info ( "Started YARN shuffle service for Spark on port {}. " + "Authentication is {}. Registered executor file is {}" , port , authEnabledString , registeredExecutorFile ) ; } catch ( Exception e ) { if ( stopOnFailure ) { throw e ; } else { noteFailure ( e ) ; } } }
Start the shuffle server with the given configuration .