Wraps a scala/java spark user defined function
514061
def wrap_spark_sql_udf(self, name, package_name=None, object_name=None, java_class_instance=None, doc=""):
        
        def _(*cols):
            jcontainer = self.get_java_container(package_name=package_name, object_name=object_name, java_class_instance=java_class_instance)
            # Ensure that your argument is a column
            function = getattr(jcontainer, name)
            judf = function()
            jc = judf.apply(self.to_scala_seq([_to_java_column(c) for c in cols]))
            return Column(jc)
        _.__name__ = name
        _.__doc__ = doc
        return _
535722
def _create_window_function(name, doc=''):
    
    def _():
        sc = SparkContext._active_spark_context
        jc = getattr(sc._jvm.functions, name)()
        return Column(jc)
    _.__name__ = name
    _.__doc__ = 'Window function: ' + doc
    return _
253615
def serialized(func):
    """"""

    @wraps(func)
    def wrapper(*args, **kwargs):
        return '\n'.join(
            ' '.join(parts) if isinstance(parts, tuple) else parts
            for parts in func(*args, **kwargs)
        )

    return wrapper
1292300
def consume_function(input = @tokens)
      function = {
        :name   => input.current[:value],
        :value  => [],
        :tokens => [input.current] # Non-standard, used for serialization.
      }

      function[:tokens].concat(input.collect {
        while token = input.consume
          case token[:node]
          when :')'
            break

          # Non-standard.
          when :comment
            next

          else
            input.reconsume
            function[:value] << consume_component_value(input)
          end
        end
      })

      create_node(:function, function)
    end
887650
def registerJavaFunction(self, name, javaClassName, returnType=None):
        
        warnings.warn(
            "Deprecated in 2.3.0. Use spark.udf.registerJavaFunction instead.",
            DeprecationWarning)
        return self.sparkSession.udf.registerJavaFunction(name, javaClassName, returnType)
322938
def foreach(self, func):
        
        if self.mode == 'spark':
            self.values.tordd().map(lambda kv: (kv[0][0], kv[1])).foreach(func)
        else:
            [func(kv) for kv in enumerate(self.values)]
80825
def _wrapper(func, *vect_args, **vect_kwargs):
        """"""
        if not hasattr(func, '__name__'):
            # Set name if not available. Happens if func is actually a function
            func.__name__ = '{}.__call__'.format(func.__class__.__name__)

        return wraps(func)(_NumpyVectorizeWrapper(func, *vect_args,
                                                  **vect_kwargs))
824768
def _create_function_over_column(name, doc=""):
    
    def _(col):
        sc = SparkContext._active_spark_context
        jc = getattr(sc._jvm.functions, name)(_to_java_column(col))
        return Column(jc)
    _.__name__ = name
    _.__doc__ = doc
    return _
305851
def transform(self, dstreams, transformFunc):
        
        jdstreams = [d._jdstream for d in dstreams]
        # change the final serializer to sc.serializer
        func = TransformFunction(self._sc,
                                 lambda t, *rdds: transformFunc(rdds),
                                 *[d._jrdd_deserializer for d in dstreams])
        jfunc = self._jvm.TransformFunction(func)
        jdstream = self._jssc.transform(jdstreams, jfunc)
        return DStream(jdstream, self, self._sc.serializer)
831039
function wrapFunctionCond(_typeVarMap, index, value) {
      var expType = typeInfo.types[index];
      if (expType.type !== FUNCTION) return value;

      //  checkValue :: (TypeVarMap, Integer, String, a) -> Either (() -> Error) TypeVarMap
      function checkValue(typeVarMap, index, k, x) {
        var propPath = [k];
        var t = expType.types[k];
        return (
          t.type === VARIABLE ?
            Z.chain (
              function(typeVarMap) {
                return isEmpty (typeVarMap[t.name].types) ?
                  Left (function() {
                    return typeVarConstraintViolation (
                      env,
                      typeInfo,
                      index,
                      propPath,
                      typeVarMap[t.name].valuesByPath
                    );
                  }) :
                  Right (typeVarMap);
              },
              Right (updateTypeVarMap (env,
                                       typeVarMap,
                                       t,
                                       index,
                                       propPath,
                                       [x]))
            ) :
          // else
            Z.map (
              function(r) { return r.typeVarMap; },
              satisfactoryTypes (env,
                                 typeInfo,
                                 typeVarMap,
                                 t,
                                 index,
                                 propPath,
                                 [x])
            )
        );
      }

      var typeVarMap = _typeVarMap;
      return function(x) {
        if (arguments.length !== expType.arity - 1) {
          throw invalidArgumentsLength (typeInfo,
                                        index,
                                        expType.arity - 1,
                                        slice.call (arguments));
        }

        var args = arguments;
        typeVarMap = assertRight (
          (init (expType.keys)).reduce (function(either, k, idx) {
            var arg = args[idx];
            return Z.chain (function(typeVarMap) {
              return checkValue (typeVarMap, index, k, arg);
            }, either);
          }, Right (typeVarMap))
        );

        var output = value.apply (this, arguments);
        var k = last (expType.keys);
        typeVarMap = assertRight (checkValue (typeVarMap, index, k, output));
        return output;
      };
    }
1193752
function run(spark) {
    var start = new Date().getTime();

    function parseRating(str) {
        var fields = str.split(",");
        if (fields.length != 4) {
            throw("Each line must contain 4 fields");
        }
        var userId = parseInt(fields[0]);
        var movieId = parseInt(fields[1]);
        var rating = parseFloat(fields[2]);
        var timestamp = parseInt(fields[3]);

        // return a plain old JSON object
        return {userId:userId, movieId:movieId, rating:rating, timestamp:timestamp};
    }

    function parseMovie(str) {
        // movieId,title,genres
        var fields = str.split(",");
        // print("str " + str + " length " + fields.length)
        var movieId = parseInt(fields[0]);
        var title = fields[1];
        var genres = fields[2]; // Adventure|Animation|Children|Comedy|Fantasy
        if (fields.length >3) {
            /*
            deal with movie titles that have a comma in them
             "American President, The (1995)"
             */
            title = fields[1] + fields[2];
            genres = fields[3]
        }

        // return a plain old JSON object
        return {movieId:movieId, title: title, genres: genres};
    }

    var Encoders = require('eclairjs/sql/Encoders');
    var Dataset = require('eclairjs/sql/Dataset');
    var DataTypes = require('eclairjs/sql/types/DataTypes');
    var ALS = require('eclairjs/ml/recommendation/ALS');
    var RegressionEvaluator = require('eclairjs/ml/evaluation/RegressionEvaluator');

    /*
    load ratings small dataset
     */
    var small_ratings_raw_data = spark.read().textFile('examples/data/mllib/ml-latest-small/ratings.csv').toRDD();
    var small_ratings_raw_data_header = small_ratings_raw_data.take(1)[0];
    var small_ratings_data_rdd = small_ratings_raw_data.filter(function(line, small_ratings_raw_data_header) {
            // filters out the header
            return line != small_ratings_raw_data_header;
        }, [small_ratings_raw_data_header])
        .map(parseRating);
    //print('small_ratings_data_rdd'+JSON.stringify(small_ratings_data.take(3)));
    var small_ratings_data = small_ratings_data_rdd.collect();

    var ratingEncoder=Encoders.json({userId:"Integer", movieId:"Integer", rating:"Float", timestamp:"Double"});
    var small_ratings_dataset = spark.createDatasetFromJson(small_ratings_data, ratingEncoder);

    //print("small_ratings_dataset " + JSON.stringify(small_ratings_dataset.take(3)));
    small_ratings_dataset.show(5);
    small_ratings_dataset.printSchema();

    /*
    load movie dataset small
     */

    var small_movies_raw_data = spark.read().textFile('examples/data/mllib/ml-latest-small/movies.csv').toRDD();
    var small_movies_raw_data_header = small_movies_raw_data.take(1)[0];
    var small_movies_data_rdd = small_movies_raw_data.filter(function(line, small_movies_raw_data_header) {
            // filters out the header
            return line != small_movies_raw_data_header;
        }, [small_movies_raw_data_header])
        .map(parseMovie).cache();
    var small_movies_data = small_movies_data_rdd.collect();

    var movieEncoder=Encoders.json({movieId:"Integer", title:"String", genres:"String"})
    var small_movie_dataset = spark.createDatasetFromJson(small_movies_data, movieEncoder);

    //print("small_movie_dataset " + JSON.stringify(small_movie_dataset.take(3)));
    small_movie_dataset.show(5);
    small_movie_dataset.printSchema();

    //var small_movies_titles = small_movie_dataframe.select("movieId", "title");
    var small_movies_titles = small_movie_dataset.select("movieId", "title");
    print("small_movies_titles " + JSON.stringify(small_movies_titles.take(3)));

    var seed = 0;
    //var split = small_ratings_dataset.randomSplit([0.6, 0.2, 0.2], seed)
    var split = small_ratings_dataset.randomSplit([0.8, 0.2, 0.2]);
    var training_DF = split[0];
    //var validation_DF = split[2];
    var test_DF = split[1];

   /* var  validation_for_predict_DF = validation_DF.select("userId", "movieId");
     print("validation_for_predict_DF " + JSON.stringify(validation_for_predict_DF.take(3)));

    var test_for_predict_DF = test_DF.select("userId", "movieId");

    print("test_for_predict_DF " + JSON.stringify(test_for_predict_DF.take(3)));
*/
    seed = 5
    var iterations = 10
    var regularization_parameter = 0.1
    var ranks = [4, 8, 12];
    var errors = [0, 0, 0];
    var err = 0
    var tolerance = 0.02

    var min_error = Number.POSITIVE_INFINITY
    var best_rank = -1
    var best_iteration = -1
    var blocks = -1;
    var lambda = regularization_parameter;

    ranks.forEach(function(rank) {
       //print("rank " + rank);
       // var model = ALS.train(training_RDD, rank, iterations, regularization_parameter, blocks, seed);
        // Build the recommendation model using ALS on the training data
        var als = new ALS()
           // .setRank(rank)
            .setMaxIter(5)
            .setRegParam(0.01)
            //.setNumBlocks(blocks)
           // .setSeed(seed)
            .setUserCol("userId")
            .setItemCol("movieId")
            .setRatingCol("rating");
        var model = als.fit(training_DF);
        //print(JSON.stringify(model.userFeatures()));
        // Evaluate the model by computing the RMSE on the test data
        var rawPredictions = model.transform(test_DF);
        var predictions = rawPredictions
            .withColumn("rating", rawPredictions.col("rating").cast(DataTypes.DoubleType))
            .withColumn("prediction", rawPredictions.col("prediction").cast(DataTypes.DoubleType));
        //print("rawPredictions " + JSON.stringify(rawPredictions));
        var evaluator = new RegressionEvaluator()
            .setMetricName("rmse")
            .setLabelCol("rating")
            .setPredictionCol("prediction");
        /*
        FIXME we are blocked until this issue is resolved
         RegressionEvaluator returns NaN for ALS in Spark ml
         https://issues.apache.org/jira/browse/SPARK-14489?jql=text%20~%20%22RegressionEvaluator%22
         */
        var rmse = evaluator.evaluate(predictions);

        errors[err] = rmse;
        err += 1;
        print("For rank " +  rank + " the RMSE is " +  rmse);
        if (rmse < min_error) {
            min_error = rmse;
            best_rank = rank;
        }

    });
    print("The best model was trained with rank " +best_rank);



    var end = new Date().getTime();
    var time = end - start;
    print('Execution time: ' + time + " milliseconds");


}
132007
def wraps_with_context(func, context):
    """"""
    wrapped = functools.partial(func, context)
    wrapped = functools.wraps(func)(wrapped)
    if asyncio.iscoroutinefunction(func):
        wrapped = asyncio.coroutine(wrapped)
    return wrapped
1248897
def expr(str):
    
    sc = SparkContext._active_spark_context
    return Column(sc._jvm.functions.expr(str))
189781
def scalarDecorator(func):
    """"""
    @wraps(func)
    def scalar_wrapper(*args,**kwargs):
        if nu.array(args[0]).shape == ():
            scalarOut= True
            newargs= ()
            for ii in range(len(args)):
                newargs= newargs+(nu.array([args[ii]]),)
            args= newargs
        else:
            scalarOut= False
        result= func(*args,**kwargs)
        if scalarOut:
            out= ()
            for ii in range(result.shape[1]):
                out= out+(result[0,ii],)
            return out
        else:
            return result
    return scalar_wrapper
376806
def callJavaFunc(sc, func, *args):
    
    args = [_py2java(sc, a) for a in args]
    return _java2py(sc, func(*args))
781166
def SpiceUDSTEP(f):
    

    @functools.wraps(f)
    def wrapping_udstep(x, value):
        result = f(x)
        value[0] = c_double(result)

    return UDSTEP(wrapping_udstep)
641097
def identical_signature_wrapper(original_function, wrapped_function):
    '''
    
    '''
    context = {'__wrapped__': wrapped_function}
    function_def = compile(
        'def {0}({1}):\n'
        '    return __wrapped__({2})'.format(
            # Keep the original function name
            original_function.__name__,
            # The function signature including defaults, i.e., 'timeout=1'
            inspect.formatargspec(
                *salt.utils.args.get_function_argspec(original_function)
            )[1:-1],
            # The function signature without the defaults
            inspect.formatargspec(
                formatvalue=lambda val: '',
                *salt.utils.args.get_function_argspec(original_function)
            )[1:-1]
        ),
        '<string>',
        'exec'
    )
    six.exec_(function_def, context)
    return wraps(original_function)(context[original_function.__name__])