小编oth*_*r15的帖子

我正在使用Scala并希望构建自己的DataFrame函数.例如,我想将列视为数组,遍历每个元素并进行计算.

首先,我正在尝试实现自己的getMax方法.因此,列x将具有值[3,8,2,5,9],并且该方法的预期输出将为9.

这是Scala中的样子

def getMax(inputArray: Array[Int]): Int = {
   var maxValue = inputArray(0)
   for (i <- 1 until inputArray.length if inputArray(i) > maxValue) {
     maxValue = inputArray(i)
   }
   maxValue
}

这是我到目前为止,并得到这个错误

"value length is not a member of org.apache.spark.sql.column", 

而且我不知道如何遍历该列.

def getMax(col: Column): Column = {
var maxValue = col(0)
for (i <- 1 until col.length if col(i) > maxValue){
    maxValue = col(i)
}
maxValue

}

一旦我能够实现自己的方法,我将创建一个列函数

val value_max:org.apache.spark.sql.Column=getMax(df.col(“value”)).as(“value_max”)

然后我希望能够在SQL语句中使用它,例如

val sample = sqlContext.sql("SELECT value_max(x) FROM table")

给定输入栏[3,8,2,5,9],预期输出为9

我正在关注另一个线程Spark …

我在理解Spark的交叉验证方面遇到了一些麻烦.我见过的任何一个例子都用它来进行参数调整,但我认为它只会进行常规的K折交叉验证吗？

我想要做的是执行k折交叉验证,其中k = 5.我想获得每个结果的准确性,然后获得平均准确度.在scikit中学习这是怎么做的,分数会给你每个折叠的结果,然后你可以使用scores.mean()

scores = cross_val_score(classifier, y, x, cv=5, scoring='accuracy')

这就是我在Spark中的做法,paramGridBuilder是空的,因为我不想输入任何参数.

val paramGrid = new ParamGridBuilder().build()
val evaluator = new MulticlassClassificationEvaluator()
  evaluator.setLabelCol("label")
  evaluator.setPredictionCol("prediction")
evaluator.setMetricName("precision")


val crossval = new CrossValidator()
crossval.setEstimator(classifier)
crossval.setEvaluator(evaluator) 
crossval.setEstimatorParamMaps(paramGrid)
crossval.setNumFolds(5)


val modelCV = crossval.fit(df4)
val chk = modelCV.avgMetrics

这和scikit学习实现的做法是一样的吗？为什么这些示例在进行交叉验证时会使用培训/测试数据？

如何交叉验证RandomForest模型？

https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/ml/ModelSelectionViaCrossValidationExample.scala

我正在尝试创建一个udf,用0替换列中的负值.

我的数据帧叫做df,包含一个名为avg_x的列.这是我创建udf的代码

val noNegative = udf {(avg_acc_x: Double) => if(avg_acc_x < 0) 0 else "avg_acc_x"}

我收到这个错误

java.lang.UnsupportedOperationException: Schema for type Any is not supported

df.printSchema返回

|-- avg_acc_x: double (nullable = false) 

所以我不明白为什么会出现这种错误？

我正在使用RandomForest.featureImportances但我不理解输出结果.

我有12个功能,这是我得到的输出.

我知道这可能不是一个特定于apache-spark的问题,但我无法找到解释输出的任何地方.

// org.apache.spark.mllib.linalg.Vector = (12,[0,1,2,3,4,5,6,7,8,9,10,11],
 [0.1956128039688559,0.06863606797951556,0.11302128590305296,0.091986700351889,0.03430651625283274,0.05975817050022879,0.06929766152519388,0.052654922125615934,0.06437052114945474,0.1601713590349946,0.0324327322375338,0.057751258970832206])

我正在为Pyspark中的分类器准备输入数据.我一直在SparkSQL中使用聚合函数来提取平均值和方差等功能.这些按活动,名称和窗口分组.通过将unix时间戳除以10000来分解为10秒的时间窗来计算窗口.

sample = sqlContext.sql("SELECT activity, name, window, avg(acc_x) as avgX , variance(acc_x) as varX FROM data  GROUP BY activity,name,window ORDER BY activity,name,window")

结果看起来像

Activity  Name         Window       AvgX       VarX
Walk    accelerometer  95875        2.0          1.0

我现在要做的是计算X中每个点的平均斜率.

为此,我需要时间戳,窗口和X.我已经在Python中实现了逻辑,使用数组,这就是它的样子 - 计算每个点之间的斜率,然后获得平均斜率.理想情况下,我想在UDAF中这样做,Pyspark尚不支持.(看起来像这样,如果下面的函数被称为斜率.那么在sql中你可以做到slope(timestamp, X) as avgSlopeX

编辑 - 更改输入,使其更清晰. 所以,我正在做的就是计算每个点之间的斜率,然后返回该窗口中斜率的平均值.所以,当我得到每个窗口的平均值和方差时,我也希望得到平均斜率.

#sample input
timestamp = [1464703425544,1464703426534,1464703427551,1464703428587,1464703429512,1464703430493,1464703431505,1464703432543,1464703433513,1464703434529]

values = [1021.31,1021.26,1021.19,1021.19,1021.1,1021.1,1021.1, 1021.05,1021.02]

i = 0; 
slope = 0.0;
totalSlope = 0.0;

while (i < len(timestamp) - 1):
    y2 = values[i+1];
    y1 = values[i];

    x2 = timestamp[i + 1]; …

我正在使用Scala并使用StringIndexer为我的训练集中的每个类别分配索引.它根据每个类别的频率分配指数.

的问题是,在我的测试数据,类别的频率是不同的,因此StringIndexer分配不同的索引的类别,这阻止我正确地评估所述模型(随机森林).

我正在以完全相同的方式处理训练/测试数据,并且不保存模型.

我尝试手动创建标签(通过获取类别的索引),但得到此错误

java.lang.IllegalArgumentException: RandomForestClassifier was given input with invalid label column label, without the number of classes specified. See StringIndexer.

似乎我必须使用StringIndexer,那么如何确保我用于测试的未来数据集以与训练集相同的方式索引类别？

编辑添加我尝试过的解决方法的代码

这就是数据框的样子,称之为mydata

+--------+-----+---------+---------+
|category|label|        x|        y|
+--------+-----+---------+---------+
| a|      0.0|  -0.166992|-0.256348|
| b|      1.0|  -0.179199| -0.22998|
| c|      2.0|  -0.172119|-0.105713|
| d|      3.0|  -0.064209| 0.050293|

我使用矢量汇编程序来准备功能

val assembler = new VectorAssembler().setInputCols(Array("x, y")).setOutputCol("features")

使用上面的汇编程序转换mydata,它使用features列

val predValues = assembler.transform(mydata)

因此该模型需要2列,功能和标签.所以我想用我自己的标签.我从predvalues中选择了特征

 val features = sqlContext.sql("SELECT features from predValues")

并从我的df中选择标签

 val labelDF = sqlContext.sql("SELECT label FROM filterFeaturesOnly")

然后将两者连接在一起,这样我就可以将功能和标签传递给模型

val featuresAndLabels = …