There is nothing strange about the output. Your vector seems to have lots of zero elements thus spark used it’s sparse representation. To explain further : It seems like your vector is composed of 18 elements (dimension). This indices [0,1,6,9,14,17] from the vector contains non zero elements which are in order [17.0,15.0,3.0,1.0,4.0,2.0] Sparse Vector representation … Read more
User defined functions are defined for up to 22 parameters. Only udf helper is define for at most 10 arguments. To handle functions with larger number of parameters you can use org.apache.spark.sql.UDFRegistration. For example val dummy = (( x0: Int, x1: Int, x2: Int, x3: Int, x4: Int, x5: Int, x6: Int, x7: Int, x8: … Read more
You can simply foldLeft over the Array of columns: val transformed: DataFrame = df.columns.foldLeft(df)((df, arg) => str(arg, df)) Still, I will argue that it is not a good approach. Since src discards StringIndexerModel it cannot be used when you get new data. Because of that I would recommend using Pipeline: import org.apache.spark.ml.Pipeline val transformers: Array[org.apache.spark.ml.PipelineStage] … Read more
There are two basic conditions under which MatrixFactorizationMode.predictAll may return a RDD with lower number of items than the input: user is missing in the training set. product is missing in the training set. You can easily reproduce this behavior and check that it is is not dependent on the way how RDD has been … Read more
Starting with Spark 1.0 there are two methods you can use to solve this easily: RDD.zipWithIndex is just like Seq.zipWithIndex, it adds contiguous (Long) numbers. This needs to count the elements in each partition first, so your input will be evaluated twice. Cache your input RDD if you want to use this. RDD.zipWithUniqueId also gives … Read more
The problem is that LinearRegressionWithSGD uses stochastic gradient descent (SGD) to optimize the weight vector of your linear model. SGD is really sensitive to the provided stepSize which is used to update the intermediate solution. What SGD does is to calculate the gradient g of the cost function given a sample of the input points … Read more
DataFrame.rdd returns RDD[Row] not RDD[(T, U)]. You have to pattern match the Row or directly extract interesting parts. ml Vector used with Datasets since Spark 2.0 is not the same as mllib Vector use by old API. You have to convert it to use with IndexedRowMatrix. Index has to be Long not string. import org.apache.spark.sql.Row … Read more
This is actually a good question. I work with mllib regularly and did not realize these basic linear algebra operations are not easily accessible. The point is that the underlying breeze vectors have all of the linear algebra manipulations you would expect – including of course basic element wise addition that you specifically mentioned. However … Read more
One way to save a model to HDFS is as following: // persist model to HDFS sc.parallelize(Seq(model), 1).saveAsObjectFile(“hdfs:///user/root/linReg.model”) Saved model can then be loaded as: val linRegModel = sc.objectFile[LinearRegressionModel](“linReg.model”).first() For more details see (ref)
As of Spark 2.3.0 there’s a much, much better way to do this. Simply extend DefaultParamsWritable and DefaultParamsReadable and your class will automatically have write and read methods that will save your params and will be used by the PipelineModel serialization system. The docs were not really clear, and I had to do a bit … Read more