标签: apache-spark-mllib

我有大约5000行和950列的csv文件.首先我将它加载到DataFrame:

val data = sqlContext.read
  .format(csvFormat)
  .option("header", "true")
  .option("inferSchema", "true")
  .load(file)
  .cache()

之后我搜索所有字符串列

val featuresToIndex = data.schema
  .filter(_.dataType == StringType)
  .map(field => field.name)

并想要索引它们.为此,我为每个字符串列创建索引器

val stringIndexers = featuresToIndex.map(colName =>
  new StringIndexer()
    .setInputCol(colName)
    .setOutputCol(colName + "Indexed"))

并创建管道

val pipeline = new Pipeline().setStages(stringIndexers.toArray)

但是当我尝试用这个管道转换我的初始数据帧时

val indexedDf = pipeline.fit(data).transform(data)

我得到StackOverflowError

16/07/05 16:55:12 INFO DAGScheduler: Job 4 finished: countByValue at StringIndexer.scala:86, took 7.882774 s
Exception in thread "main" java.lang.StackOverflowError
at scala.collection.immutable.Set$Set1.contains(Set.scala:84)
at scala.collection.immutable.Set$Set1.$plus(Set.scala:86)
at scala.collection.immutable.Set$Set1.$plus(Set.scala:81)
at scala.collection.mutable.SetBuilder.$plus$eq(SetBuilder.scala:22)
at scala.collection.mutable.SetBuilder.$plus$eq(SetBuilder.scala:20)
at scala.collection.generic.Growable$class.loop$1(Growable.scala:53)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:57)
at …

我正在尝试使用Spark的MLLib构建一个NaiveBayes分类器,它将一组文档作为输入.

我想把一些东西作为特征(即作者,显式标签,隐式关键字,类别),但看文档似乎LabeledPoint只包含双打,即它看起来像LabeledPoint[Double, List[Pair[Double,Double]].

相反,我从其余代码输出的内容就像是LabeledPoint[Double, List[Pair[String,Double]].

我可以编造自己的转换,但看起来很奇怪.我怎么用MLLib来处理这个问题？

我相信答案是在HashingTF课堂上(即散列功能),但我不明白它是如何工作的,它似乎需要某种容量值,但我的关键词和主题列表实际上是无限的(或更好,未知一开始).

我试图找出是否有可能在Apache Spark中使用MLlib对数据进行"增量训练".

我的平台是Prediction IO,它基本上是Spark(MLlib),HBase,ElasticSearch和其他一些Restful部件的包装器.

在我的应用数据中,"事件"是实时插入的,但为了获得更新的预测结果,我需要"pio train"和"pio deploy".这需要一些时间,服务器在重新部署期间会脱机.

我想弄清楚我是否可以在"预测"阶段进行增量训练,但找不到答案.

我一直在寻找的例子星火网站Word2Vec的:

val input = sc.textFile("text8").map(line => line.split(" ").toSeq)

val word2vec = new Word2Vec()

val model = word2vec.fit(input)

val synonyms = model.findSynonyms("country name here", 40)

我如何做有趣的矢量,如国王 - 男人+女人=女王.我可以使用model.getVectors,但不知道如何继续进行.

我正在编写一个spark应用程序,并希望在MLlib中使用算法.在API文档中,我发现了同一算法的两个不同的类.

例如,org.apache.spark.ml.classification中的一个LogisticRegression也是org.apache.spark.mllib.classification中的LogisticRegressionwithSGD.

我能找到的唯一区别是org.apache.spark.ml中的那个继承自Estimator并且能够用于交叉验证.我很困惑,他们被放在不同的包装中.有没有人知道它的原因？谢谢!

我HashingTF在Spark中创建了Term Frequency .我已经tf.transform为每个单词使用了术语频率.

但结果以这种格式显示.

[<hashIndexofHashBucketofWord1>,<hashIndexofHashBucketofWord2> ...]
,[termFrequencyofWord1, termFrequencyOfWord2 ....]

例如:

(1048576,[105,3116],[1.0,2.0])

我能够使用哈希桶获取索引tf.indexOf("word").

但是,我怎样才能使用索引得到这个词？

我有一个名为ab的DataFrame .当我将ab指定为StringIndexer的输入列名时,AnalysisException的消息"无法解析'ab'给定输入列ab".我正在使用Spark 1.6.0.

我知道旧版本的Spark可能在列名中遇到点问题,但在更新版本中,可以在Spark shell和SQL查询中使用反引号.例如,这是解决另一个问题,如何在Spark SQL中使用连字符转义列名.其中一些问题是SPARK-6898报告 的,列名中的特殊字符被破坏,但是在1.4.0中得到了解决.

这是一个最小的例子和堆栈跟踪:

public class SparkMLDotColumn {
    public static void main(String[] args) {
        // Get the contexts
        SparkConf conf = new SparkConf()
                .setMaster("local[*]")
                .setAppName("test")
                .set("spark.ui.enabled", "false"); // http://permalink.gmane.org/gmane.comp.lang.scala.spark.user/21385
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sparkContext);

        // Create a schema with a single string column named "a.b"
        StructType schema = new StructType(new StructField[] {
                DataTypes.createStructField("a.b", DataTypes.StringType, false)
        });

        // Create …

我试图从a中取列DataFrame并将其转换为RDD[Vector].

问题是我的名称中有一个带"dot"的列作为以下数据集:

"col0.1","col1.2","col2.3","col3.4"
1,2,3,4
10,12,15,3
1,12,10,5

这就是我正在做的事情:

val df = spark.read.format("csv").options(Map("header" -> "true", "inferSchema" -> "true")).load("C:/Users/mhattabi/Desktop/donnee/test.txt")
val column=df.columns.map(c=>s"`${c}`")
val rows = new VectorAssembler().setInputCols(column).setOutputCol("vs")
  .transform(df)
  .select("vs")
  .rdd
val data =rows.map(_.getAs[org.apache.spark.ml.linalg.Vector](0))
  .map(org.apache.spark.mllib.linalg.Vectors.fromML)

val mat: RowMatrix = new RowMatrix(data)
//// Compute the top 5 singular values and corresponding singular vectors.
val svd: SingularValueDecomposition[RowMatrix, Matrix] = mat.computeSVD(mat.numCols().toInt, computeU = true)
val U: RowMatrix = svd.U  // The U factor is a RowMatrix.
val s: Vector = svd.s  // The singular …

我想知道使用 Apache Spark 2.4.5 和 PySpark (Python) 评估拟合二进制分类模型的最佳方法是什么。我想考虑不同的指标，例如准确率、准确率、召回率、auc 和 f1 分数。

让我们假设给出以下内容：

# pyspark.sql.dataframe.DataFrame in VectorAssembler format containing two columns: target and features
# DataFrame we want to evaluate
df

# Fitted pyspark.ml.tuning.TrainValidationSplitModel (any arbitrary ml algorithm)
model

1. 选项

无论BinaryClassificationEvaluator也不MulticlassClassificationEvaluator可以计算出自己的上述所有指标。因此，我们使用两个评估器。

from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator

# Create both evaluators
evaluatorMulti = MulticlassClassificationEvaluator(labelCol="target", predictionCol="prediction")
evaluator = BinaryClassificationEvaluator(labelCol="target", rawPredictionCol="prediction", metricName='areaUnderROC')

# Make predicitons
predictionAndTarget = model.transform(df).select("target", "prediction")

# Get metrics
acc = evaluatorMulti.evaluate(predictionAndTarget, {evaluatorMulti.metricName: "accuracy"})
f1 = …

我有一个数据帧gi_man_df,其中group可以是n:

+------------------+-----------------+--------+--------------+
|           group  |           number|rand_int|   rand_double|
+------------------+-----------------+--------+--------------+
|          'GI_MAN'|                7|       3|         124.2|
|          'GI_MAN'|                7|      10|        121.15|
|          'GI_MAN'|                7|      11|         129.0|
|          'GI_MAN'|                7|      12|         125.0|
|          'GI_MAN'|                7|      13|         125.0|
|          'GI_MAN'|                7|      21|         127.0|
|          'GI_MAN'|                7|      22|         126.0|
+------------------+-----------------+--------+--------------+

我期待一个numpy nd_array,即gi_man_array:

[[[124.2],[121.15],[129.0],[125.0],[125.0],[127.0],[126.0]]]

应用pivot后rand_double的值.

我尝试了以下两种方法:首先
:我按如下方式转动gi_man_df:

gi_man_pivot = gi_man_df.groupBy("number").pivot('rand_int').sum("rand_double")

我得到的输出是:

Row(number=7, group=u'GI_MAN', 3=124.2, 10=121.15, 11=129.0, 12=125.0, 13=125.0, 21=127.0, 23=126.0)

但这里的问题是获得所需的输出,我无法将其转换为矩阵然后再转换为numpy数组.

SECOND: 我使用以下方法在数据框中创建了向量:

assembler = VectorAssembler(inputCols=["rand_double"],outputCol="rand_double_vector")

gi_man_vector = assembler.transform(gi_man_df)
gi_man_vector.show(7) …