Merge branch 'master' of https://github.com/EclairJS/eclairjs-node

Author: Jackie Barbetta

lib/Accumulable.js

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2015 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+var Utils = require('./utils.js');
+
+/**
+ * A data type that can be accumulated, ie has an commutative and associative "add" operation,
+ * but where the result type, `R`, may be different from the element type being added, `T`.
+ *
+ * You must define how to add data, and how to merge two of these together.  For some data types,
+ * such as a counter, these might be the same operation. In that case, you can use the simpler
+ * {@link Accumulator}. They won't always be the same, though -- e.g., imagine you are
+ * accumulating a set. You will add items to the set, and you will union two sets together.
+ *
+ * @classdesc
+ * @constructor
+ */
+/*
+ * NOTE for now EclairJS will only support floats and int types
+ *
+ *
+ */
+function Accumulable(kernelP, refIdP) {
+  this.kernelP = kernelP;
+  this.refIdP = refIdP;
+}
+
+/**
+ * Add more data to this accumulator / accumulable
+ * @param {object} term  the data to add
+ * @returns {Promise.<Void>} A Promise that resolves to nothing.
+ */
+Accumulable.prototype.add = function(term) {
+  var templateStr = '{{inRefId}}.add({{term}});';
+  return Utils.generateVoidPromise(this, templateStr, {term: Utils.prepForReplacement(term)});
+};
+
+
+/**
+ * Merge two accumulable objects together
+ *
+ * Normally, a user will not want to use this version, but will instead call `add`.
+ * @param {object} term  the other `R` that will get merged with this
+ * @returns {Promise.<Void>} A Promise that resolves to nothing.
+ */
+Accumulable.prototype.merge = function(term) {
+  var templateStr = '{{inRefId}}.merge({{term}});';
+  return Utils.generateVoidPromise(this, templateStr, {term: Utils.prepForReplacement(term)});
+};
+
+/**
+ * Access the accumulator's current value; only allowed on master.
+ * @returns {Promise.<object>}
+ */
+Accumulable.prototype.value = function() {
+  function _resolve(result, resolve, reject) {
+    resolve(parseFloat(result));
+  }
+
+  var templateStr = '{{inRefId}}.value();';
+
+  return Utils.generateResultPromise(this, templateStr, null, _resolve);
+};
+
+/**
+ * Get the current value of this accumulator from within a task.
+ *
+ * This is NOT the global value of the accumulator.  To get the global value after a
+ * completed operation on the dataset, call `value`.
+ *
+ * The typical use of this method is to directly mutate the local value, eg., to add
+ * an element to a Set.
+ * @returns {Promise.<object>}
+ */
+Accumulable.prototype.localValue = function() {
+  function _resolve(result, resolve, reject) {
+    resolve(parseFloat(result));
+  }
+
+  var templateStr = '{{inRefId}}.localValue();';
+
+  return Utils.generateResultPromise(this, templateStr, null, _resolve);
+};
+
+/**
+ * Set the accumulator's value; only allowed on master
+ * @param {object}
+ * @returns {Promise.<Void>} A Promise that resolves to nothing.
+ */
+Accumulable.prototype.setValue = function(newValue) {
+  var templateStr = '{{inRefId}}.setValue({{newValue}});';
+
+  return Utils.generateVoidPromise(this, templateStr , {newValue: Utils.prepForReplacement(newValue)});
+};
+
+/**
+ * @returns {Promise.<string>}
+ */
+Accumulable.prototype.toString = function() {
+  var templateStr = '{{inRefId}}.toString();';
+  return Utils.generateResultPromise(this, templateStr);
+};
+
+module.exports = Accumulable;

lib/AccumulableParam.js

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2015 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+var Utils = require('./utils.js');
+
+var kernelP;
+
+/**
+ * Helper object defining how to accumulate values of a particular type. An implicit
+ * AccumulableParam needs to be available when you create {@link Accumulable}s of a specific type.
+ *
+ * @classdesc
+ * @constructor
+ */
+function AccumulableParam() {
+}
+
+/**
+ * Add additional data to the accumulator value. Is allowed to modify and return `r`
+ * for efficiency (to avoid allocating objects).
+ *
+ * @param {object} r  the current value of the accumulator
+ * @param {object} t  the data to be added to the accumulator
+ * @returns {object}  the new value of the accumulator
+ * @returns {Promise.<object>} Promise that resolves to the new value of the accumulator
+ */
+AccumulableParam.prototype.addAccumulator = function(r, t) {
+  var templateStr = '{{inRefId}}.addAccumulator({{r}}, {{t}});';
+
+  return Utils.generateResultPromise(this, templateStr, {r: r, t: t});
+};
+
+/**
+ * Merge two accumulated values together. Is allowed to modify and return the first value
+ * for efficiency (to avoid allocating objects).
+ *
+ * @param {object} r1  one set of accumulated data
+ * @param {object} r2  another set of accumulated data
+ * @returns {Promise.<object>} Promise that resolves to a value that has both data sets merged together
+ */
+AccumulableParam.prototype.addInPlace = function(r1,r2) {
+  var templateStr = '{{inRefId}}.addInPlace({{r1}},{{r2}});';
+
+ return Utils.generateResultPromise(this, templateStr, {r1: r1, r2: r2});
+};
+
+/**
+ * Return the "zero" (identity) value for an accumulator type, given its initial value. For
+ * example, if R was a vector of N dimensions, this would return a vector of N zeroes.
+ * @param {object}
+ * @returns {Promise.<object>}
+ */
+AccumulableParam.prototype.zero = function(initialValue) {
+  var templateStr = '{{inRefId}}.zero({{initialValue}});';
+
+  return Utils.generateResultPromise(this, templateStr, {initialValue: initialValue});
+};
+
+function FloatAccumulatorParam(kernelP, refIdP) {
+  this.kernelP = kernelP;
+  this.refIdP = refIdP;
+}
+
+FloatAccumulatorParam.prototype = AccumulableParam.prototype;
+
+
+function IntAccumulatorParam(kernelP, refIdP) {
+  this.kernelP = kernelP;
+  this.refIdP = refIdP;
+}
+
+IntAccumulatorParam.prototype = AccumulableParam.prototype;
+
+module.exports = function(kP) {
+  kernelP = kP;
+  return {
+    FloatAccumulatorParam: function() {
+      var templateStr = 'var {{refId}} = new FloatAccumulatorParam();';
+      return Utils.evaluate(kernelP, FloatAccumulatorParam, templateStr);
+    },
+
+    IntAccumulatorParam: function() {
+      var templateStr = 'var {{refId}} = new IntAccumulatorParam();';
+      return Utils.evaluate(kernelP, IntAccumulatorParam, templateStr);
+    }
+  };
+};

lib/Accumulator.js

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2015 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+var Accumulable = require('./Accumulable.js');
+
+/**
+ * A simpler value of {@link Accumulable} where the result type being accumulated is the same
+ * as the types of elements being merged, i.e. variables that are only "added" to through an
+ * associative operation and can therefore be efficiently supported in parallel. They can be used
+ * to implement counters (as in MapReduce) or sums. EclairJS supports accumulators of numeric
+ * value types.
+ *
+ * An accumulator is created from an initial value `v` by calling [[SparkContext#accumulator]].
+ * Tasks running on the cluster can then add to it using the [[Accumulable#add]].
+ * However, they cannot read its value. Only the driver program can read the accumulator's value,
+ * using its value method.
+ *
+ *
+ * @example
+ * 	var accum = sparkContext.accumulator(0);
+ *	sparkContext.parallelize([1, 2, 3, 4])
+ *				.foreach(function(x, accum) {
+ *					accum.add(x);
+ *				});
+ *	print(accum.value()); // displays 10
+ *
+ */
+
+function Accumulator(kernelP, refIdP) {
+  this.kernelP = kernelP;
+  this.refIdP = refIdP;
+}
+
+Accumulator.prototype = Accumulable.prototype;
+
+module.exports = Accumulator;

lib/SparkContext.js

@@ -46,17 +46,13 @@ var kernelP = new Promise(function(resolve, reject) {
  * @param {string} name - A name for your application, to display on the cluster web UI
  */
 function SparkContext(master, name) {
- kernel.createKernelSession(name).then(kernelPResolve).catch(kernelPReject);
-
-  var myThis = this;
+  kernel.createKernelSession(name).then(kernelPResolve).catch(kernelPReject);
 
   this.kernelP = new Promise(function(resolve, reject) {
     kernelP.then(function(kernel) {
       var templateStr = 'var jsc = new SparkContext("{{master}}", "{{name}}");';
 
       Utils.execute(kernelP, templateStr, {master: master, name: name}).then(function() {
-       // resolve(k);
-
         // Check version
         templateStr = 'jsc.version();';
         // This is somewhat ugly, since SparkContext's kernelP hasn't been resolved yet.
@@ -73,13 +69,59 @@ function SparkContext(master, name) {
   });
 }
 
+/**
+ * Create an {@link Accumulable} shared variable of the given type, to which tasks can "add" values with add.
+ * Only the master can access the accumuable's value.
+ *
+ * @param {object} initialValue
+ * @param {AccumulableParam} param
+ * @param {string} name of  the accumulator for display in Spark's web UI.
+ * @returns {Accumulable}
+ */
+SparkContext.prototype.accumulable = function() {
+  var Accumulable = require('./Accumulable.js');
+
+  if (arguments.length == 3) {
+    var templateStr = 'var {{refId}} = jsc.accumulable({{initialValue}}, {{name}}, {{param}});';
+
+    return Utils.evaluate(this.kernelP, Accumulable, templateStr, {initialValue: arguments[0], name: Utils.prepForReplacement(arguments[1]), param: Utils.prepForReplacement(arguments[2])});
+  } else {
+    var templateStr = 'var {{refId}} = jsc.accumulable({{initialValue}}, {{param}});';
+
+    return Utils.evaluate(this.kernelP, Accumulable, templateStr, {initialValue: arguments[0], param: Utils.prepForReplacement(arguments[1])});
+  }
+};
+
+/**
+ * Create an {@link Accumulator}  variable, which tasks can "add" values to using the add method.
+ * Only the master can access the accumulator's value.
+ *
+ * @param {int | float} initialValue
+ * @param {string} name of  the accumulator for display in Spark's web UI. Optional
+ * @param {AccumulableParam} param Optional defaults to FloatAccumulatorParam
+ * @returns {Accumulator}
+ */
+SparkContext.prototype.accumulator = function() {
+  var Accumulator = require('./Accumulator.js');
+
+  if (arguments.length == 3) {
+    var templateStr = 'var {{refId}} = jsc.accumulator({{initialValue}}, {{name}}, {{param}});';
+
+    return Utils.evaluate(this.kernelP, Accumulator, templateStr, {initialValue: arguments[0], name: Utils.prepForReplacement(arguments[1]), param: Utils.prepForReplacement(arguments[2])});
+  } else {
+    var templateStr = 'var {{refId}} = jsc.accumulator({{initialValue}}, {{param}});';
+
+    return Utils.evaluate(this.kernelP, Accumulator, templateStr, {initialValue: arguments[0], param: Utils.prepForReplacement(arguments[1])});
+  }
+};
+
 /**
  * Distribute a local Scala collection to form an RDD.
  * @param {array} arr
  * @returns {RDD}
  */
 SparkContext.prototype.parallelize = function(arr) {
-  var templateStr = 'var {{refId}} = jsc.parallelize("{{arr}}");';
+  var templateStr = 'var {{refId}} = jsc.parallelize([{{arr}}]);';
 
   return Utils.evaluate(this.kernelP, RDD, templateStr, {arr: arr.join(',')});
 };

lib/index.js

@@ -26,6 +26,8 @@ function EclairJS() {
   return {
     SparkContext: result[1],
 
+    AccumulableParam: require('./AccumulableParam.js')(kernelP),
+
     SQLContext: require('./sql/SQLContext.js'),
     sql: {
       functions: require('./sql/functions.js')(kernelP)

lib/sql/DataFrame.js

@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-var protocol = require('../kernel.js');
 var Utils = require('../utils.js');
 var serialize = require('../serialize.js');