How to set and manage compute context in Machine Learning Server

In Machine Learning Server, every session that loads a function library has a compute context. The default is local, available on all platforms. No action is required to use a local compute context.

This article explains how to shift script execution to a remote Hadoop or Spark cluster, for the purpose of bring calculations to the data itself and eliminating data transfer over your network. For SQL Server compute contexts, see Define and use a compute context in the SQL Server documentation.

You can create multiple compute context objects: just use them one at a time. Often, functions operate identically in local and remote context. If script execution is successful locally, you can generally expect the same results on the remote server, subject to these limitations.

Prerequisites

The remote computer must be at the same functional level as the system sending the request. A remote Machine Learning Server 9.2.1 can accept a compute context shift from another 9.2.1 interpreter on Machine Learning Server, a SQL Server 2017 Machine Learning Services instance, and for R developers on Microsoft R Client at the same functional level (3.4.1).

The following table is a recap of platform and data source requirements for each function library.

⁽¹⁾ You can load text or .xdf files locally, but be aware that code and data runs on SQL Server, which results in implicit data type conversions.

Get a compute context

As a starting point, practice the syntax for returning the local compute context. Every platform, product, and language supports the default local compute context.

# RevoScaleR is loaded automatically so no import step is required
rxGetComputeContext()

from revoscalepy import RxLocalSeq
localcc = RxLocalSeq()

For both languages, the return object should be Return object should be RxLocalSeq.

Set a remote compute context

This section uses examples to illustrate the syntax for setting compute context.

For Spark

The compute context used to distribute computations on a Hadoop Spark 2.0-2.1 cluster. For more information, see the How to use RevoScaleR with Spark.

    myHadoopCluster <- RxSpark(myHadoopCluster)

For MapReduce

The compute context used to distribute computations on a Hadoop MapReduce cluster. This compute context can be used on a node (including an edge node) of a Cloudera or Hortonworks cluster with an RHEL operating system, or a client with an SSH connection to such a cluster. For details on creating and using RxHadoopMR compute contexts, see the How to use RevoScaleR with Hadoop

    myHadoopCluster <- RxHadoopMR(myHadoopCluster)

For SQL Server (in-database)

The RxInSqlServer compute context is a special case in that it runs computations in-database, but it runs on only a single database node, so the computation is parallel, but not distributed.

For setting up a remote compute context to SQL Server, we provide an example below, but point you to Define and use a compute context in the SQL Server documentation for additional instructions.

# Requires RevoScaleR and SQL Server on same machine
connectionString <- "Server=(placeholder-server-name);Database=RevoAirlineDB;Trusted_Connection=true"

sqlQuery <- "WITH nb AS (SELECT 0 AS n UNION ALL SELECT n+1 FROM nb where n < 9) SELECT n1.n+10*n2.n+100*n3.n+1 AS n, ABS(CHECKSUM(NewId())) 

myServer <- RxComputeContext("RxInSqlServer", sqlQuery = sqlQuery, connectionString = connectionString)   
                   
rxSetComputeContext(computeContext = myServer)

Limitations of remote compute context

A primary benefit of remote computing is to perform distributed analysis using the inherent capabilities of the host platform. For a list of analytical functions that are multithreaded and cluster aware, see Running distributed analyses using RevoScaleR. As a counter point, the concept of "remote compute context" is not optimized for data manipulation and transformation workloads. As such, many data-related functions come with local compute requirements. The following table describes the exceptions to remote computing.

Set a no-wait compute context

By default, all jobs are "waiting jobs" or "blocking jobs", where control of the command prompt is not returned until the job is complete. For jobs that complete quickly, this is an appropriate choice.

For large jobs that run longer, execute the job but use a non-waiting compute context to continue working in your local session. In this case, you can specify the compute context to be non-waiting (or non-blocking), in which case an object containing information about the pending job is used to retrieve results later.

To set the compute context object to run "no wait" jobs, set the argument wait to FALSE.

    myHadoopCluster <- RxSpark(myHadoopCluster, wait=FALSE)

Another use for non-waiting compute contexts is for massively parallel jobs involving multiple clusters. You can define a non-waiting compute context on each cluster, launch all your jobs, then aggregate the results from all the jobs once they complete.

For more information, see Non-Waiting Jobs.

Retrieve cluster console output

If you want console output from each of the cluster R processes printed to your user console, specify consoleOutput=TRUE in your compute context.

    myHadoopCluster <- RxSpark(myHadoopCluster, consoleOutput=TRUE)

Update a compute context

Once you have a compute context object, modify it using the same function that originally creates it. Pass the name of the original object as its first argument, and then specify only those arguments you wish to modify as additional arguments.

For example, to change only the suppressWarning parameter of a Spark compute context myHadoopCluster from TRUE to FALSE:

    myHadoopCluster <- RxSpark(myHadoopCluster, suppressWarning=FALSE)

To list parameters and default values, use the args function with the name of the compute context constructor, for example:

	args(RxSpark)

which gives the following output:

function (object, hdfsShareDir = paste("/user/RevoShare", Sys.info()[["user"]], 
	sep = "/"), shareDir = paste("/var/RevoShare", Sys.info()[["user"]], 
	sep = "/"), clientShareDir = rxGetDefaultTmpDirByOS(), sshUsername = Sys.info()[["user"]], 
	sshHostname = NULL, sshSwitches = "", sshProfileScript = NULL, 
	sshClientDir = "", nameNode = rxGetOption("hdfsHost"), jobTrackerURL = NULL, 
	port = rxGetOption("hdfsPort"), onClusterNode = NULL, wait = TRUE, 
	numExecutors = rxGetOption("spark.numExecutors"), executorCores = rxGetOption("spark.executorCores"), 
	executorMem = rxGetOption("spark.executorMem"), driverMem = "4g", 
	executorOverheadMem = rxGetOption("spark.executorOverheadMem"), 
	extraSparkConfig = "", persistentRun = FALSE, sparkReduceMethod = "auto", 
	idleTimeout = 3600, suppressWarning = TRUE, consoleOutput = FALSE, 
	showOutputWhileWaiting = TRUE, autoCleanup = TRUE, workingDir = NULL, 
	dataPath = NULL, outDataPath = NULL, fileSystem = NULL, packagesToLoad = NULL, 
	resultsTimeout = 15, ...) 

You can temporarily modify an existing compute context and set the modified context as the current compute context by calling rxSetComputeContext. For example, if you have defined myHadoopCluster to be a waiting cluster and want to set the current compute context to be non-waiting, you can call rxSetComputeContext as follows:

	rxSetComputeContext(myHadoopCluster, wait=FALSE)

The rxSetComputeContext function returns the previous compute context, so it can be used in constructions like the following:

    oldContext <- rxSetComputeContext(myCluster, wait=FALSE)
    …
    # do some computing with a non-waiting compute context
    …
    # restore previous compute context
    rxSetComputeContext(oldContext)

You can specify the compute context by name, as we have done here, but you can also specify it by calling a compute context constructor in the call to rxSetComputeContext. For example, to return to the local sequential compute context after using a cluster context, you can call rxSetComputeContext as follows:

	rxSetComputeContext(RxLocalSeq())

In this case, you can also use the descriptive string "local" to do the same thing:

	rxSetComputeContext("local")

Create additional compute contexts

Given a set of distributed computing resources, you might want to create multiple compute context objects to vary the configuration. For example, you might have one compute context for waiting or blocking jobs and another for no-wait or non-blocking jobs. Or you might define one that uses all available nodes and another that specifies a particular set of nodes.

Because the initial specification of a compute context can be somewhat tedious, it is usually simplest to create additional compute contexts by modifying an existing compute context, in precisely the same way as we updated a compute context in the previous section. For example, suppose instead of simply modifying our existing compute context from wait=TRUE to wait=FALSE, we create a new compute context for non-waiting jobs:

	myNoWaitCluster <- RxSpark(myHadoopCluster, wait=FALSE)

See also

• Introduction to Machine Learning Server
• Distributed computing
• Compute context
• Define and use a compute context in SQL Server