org.kepler.spark.director

Class SparkEngine

java.lang.Object

org.kepler.ddp.director.DDPEngine

org.kepler.spark.director.SparkEngine

All Implemented Interfaces:

java.lang.Cloneable

public class SparkEngine
extends DDPEngine

An engine than runs workflows in Spark. This engine converts DDP pattern actors (Map, Reduce, Cross, CoGroup, and Match) and I/O actors (DDPDataSink and DDPDataSource) into a Spark job and runs it on the server.

NOTE: Only DDP pattern and I/O actors may be present in the workflow. Other actors must placed inside the composite pattern actors or in a different sub-workflow.

Author:

Daniel Crawl

Field Summary

Fields

Modifier and Type	Field and Description
StringParameter	driverMemory
StringParameter	masterHostAndPort
Parameter	numSameJVMWorkers
static java.lang.String	SPARK_ENGINE_NAME The name of the spark engine.

Fields inherited from class org.kepler.ddp.director.DDPEngine

_additionalJars, _classLoader, _configDirStr, _CONNECT_TIMEOUT, _container, _degreeOfParallelism, _director, _displayRedirectDir, _engineName, _jobArgumentsMap, _jobDir, _jobLock, _sameJVM, _serverStartStopLock, _subWorkflows, _writeSubWorkflowsToFiles

Constructor Summary

Constructors

Constructor and Description
SparkEngine(DDPDirector director)

Method Summary

Modifier and Type	Method and Description
protected void	_addParameters() Add parameters to the containing director.
protected void	_addSubModelToContract(SingleInputPatternActor pactor, Operator contract) Write the sub-workflow of a SingleInputPatternActor either to a parameter in the Spark job configuration or to a file.
protected boolean	_checkFilesBeforeStartingServer() Check the existence of the Spark assembly jar.
protected void	_executeDDPJob() Execute the DDP job.
protected SparkJob	_getModelPlan() Constructs a SparkJob from the model.
protected java.net.InetSocketAddress	_parseOutputFromStartingServer(java.io.InputStream input) Parse the output from the script that starts the server.
protected void	_removeParameters() Remove engine-specific parameters from the director.
void	attributeChanged(Attribute attribute) React to a parameter change.
java.lang.Object	clone(Workspace workspace) Clone the SparkEngine into the specified workspace.
void	preinitialize() Initialize fields from parameters.
static void	shutdownServer()
void	stop() Stop any running Spark jobs.

Methods inherited from class org.kepler.ddp.director.DDPEngine

_checkConfigDir, _checkModel, _checkModelForNonDDPActors, _checkServer, _copyParameters, _createJobDirectory, _getJarList, _getParametersInHierarchy, _setPortTypes, clone, closeAllEffigies, createEffigy, fire, getDisplayRedirectDir, getModel, getName, getServerTypes, postfire, wrapup

Methods inherited from class java.lang.Object

equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

masterHostAndPort

public StringParameter masterHostAndPort

numSameJVMWorkers

public Parameter numSameJVMWorkers

driverMemory

public StringParameter driverMemory

SPARK_ENGINE_NAME

public static final java.lang.String SPARK_ENGINE_NAME

The name of the spark engine.

See Also:

Constant Field Values

Constructor Detail

SparkEngine

public SparkEngine(DDPDirector director)
            throws IllegalActionException,
                   NameDuplicationException

Throws:

IllegalActionException

NameDuplicationException

Method Detail

attributeChanged

public void attributeChanged(Attribute attribute)
                      throws IllegalActionException

React to a parameter change.

Overrides:

attributeChanged in class DDPEngine

Throws:

IllegalActionException

clone

public java.lang.Object clone(Workspace workspace)
                       throws java.lang.CloneNotSupportedException

Clone the SparkEngine into the specified workspace.

Overrides:

clone in class DDPEngine

Parameters:

workspace - The workspace for the cloned object.

Returns:

The new SparkEngine.

Throws:

java.lang.CloneNotSupportedException - Not thrown in this base class

preinitialize

public void preinitialize()
                   throws IllegalActionException

Description copied from class: DDPEngine

Initialize fields from parameters.

Overrides:

preinitialize in class DDPEngine

Throws:

IllegalActionException

shutdownServer

public static void shutdownServer()
                           throws IllegalActionException

Throws:

IllegalActionException

stop

public void stop()
          throws IllegalActionException

Stop any running Spark jobs.

Overrides:

stop in class DDPEngine

Throws:

IllegalActionException

_checkFilesBeforeStartingServer

protected boolean _checkFilesBeforeStartingServer()
                                           throws IllegalActionException

Check the existence of the Spark assembly jar.

Overrides:

_checkFilesBeforeStartingServer in class DDPEngine

Throws:

IllegalActionException

_executeDDPJob

protected void _executeDDPJob()
                       throws IllegalActionException

Description copied from class: DDPEngine

Execute the DDP job.

Specified by:

_executeDDPJob in class DDPEngine

Throws:

IllegalActionException

_addParameters

protected void _addParameters()
                       throws IllegalActionException,
                              NameDuplicationException

Add parameters to the containing director.

Overrides:

_addParameters in class DDPEngine

Throws:

IllegalActionException

NameDuplicationException

_removeParameters

protected void _removeParameters()
                          throws IllegalActionException,
                                 NameDuplicationException

Remove engine-specific parameters from the director.

Overrides:

_removeParameters in class DDPEngine

Throws:

IllegalActionException

NameDuplicationException

_getModelPlan

protected SparkJob _getModelPlan()
                          throws IllegalActionException

Constructs a SparkJob from the model.

Throws:

IllegalActionException

_addSubModelToContract

protected void _addSubModelToContract(SingleInputPatternActor pactor,
                                      Operator contract)
                               throws IllegalActionException

Write the sub-workflow of a SingleInputPatternActor either to a parameter in the Spark job configuration or to a file.

Throws:

IllegalActionException

_parseOutputFromStartingServer

protected java.net.InetSocketAddress _parseOutputFromStartingServer(java.io.InputStream input)
                                                             throws java.io.IOException,
                                                                    IllegalActionException

Parse the output from the script that starts the server.

Overrides:

_parseOutputFromStartingServer in class DDPEngine

Returns:

If the start script specifies a server URL, returns the socket address for that URL. Otherwise, returns null.

Throws:

java.io.IOException

IllegalActionException