标签: apache-spark-ml

我正在修补PySpark文档中的一些交叉验证代码,并尝试让PySpark告诉我选择了哪个模型:

from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.mllib.linalg import Vectors
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

dataset = sqlContext.createDataFrame(
    [(Vectors.dense([0.0]), 0.0),
     (Vectors.dense([0.4]), 1.0),
     (Vectors.dense([0.5]), 0.0),
     (Vectors.dense([0.6]), 1.0),
     (Vectors.dense([1.0]), 1.0)] * 10,
    ["features", "label"])
lr = LogisticRegression()
grid = ParamGridBuilder().addGrid(lr.regParam, [0.1, 0.01, 0.001, 0.0001]).build()
evaluator = BinaryClassificationEvaluator()
cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator)
cvModel = cv.fit(dataset)

在PySpark shell中运行它,我可以得到线性回归模型的系数,但我似乎无法找到lr.regParam交叉验证程序选择的值.有任何想法吗？

In [3]: cvModel.bestModel.coefficients
Out[3]: DenseVector([3.1573])

In [4]: cvModel.bestModel.explainParams()
Out[4]: ''

In [5]: cvModel.bestModel.extractParamMap()
Out[5]: {}

In [15]: cvModel.params
Out[15]: [] …

我有一个DataFrame架构

root
 |-- label: string (nullable = true)
 |-- features: struct (nullable = true)
 |    |-- feat1: string (nullable = true)
 |    |-- feat2: string (nullable = true)
 |    |-- feat3: string (nullable = true)

同时,我能够使用过滤数据框

  val data = rawData
     .filter( !(rawData("features.feat1") <=> "100") )

我无法删除列

  val data = rawData
       .drop("features.feat1")

这是我在这里做错了吗？我也试过(不成功)做drop(rawData("features.feat1")),虽然这样做没有多大意义.

提前致谢,

尼基尔

我是Spark SQL DataFrames和ML的新手(PySpark).如何创建服装标记器,例如删除停用词并使用nltk中的某些库？我可以延长默认值吗？

谢谢.

我想知道是否有简洁的方法在pyspark中的DataFrame上运行ML(例如KMeans),如果我有多个数字列中的功能.

即在Iris数据集中:

(a1=5.1, a2=3.5, a3=1.4, a4=0.2, id=u'id_1', label=u'Iris-setosa', binomial_label=1)

我想使用KMeans而不重新创建DataSet,并将功能向量手动添加为新列,并在代码中重复硬编码原始列.

我想改进的解决方案:

from pyspark.mllib.linalg import Vectors
from pyspark.sql.types import Row
from pyspark.ml.clustering import KMeans, KMeansModel

iris = sqlContext.read.parquet("/opt/data/iris.parquet")
iris.first()
# Row(a1=5.1, a2=3.5, a3=1.4, a4=0.2, id=u'id_1', label=u'Iris-setosa', binomial_label=1)

df = iris.map(lambda r: Row(
                    id = r.id,
                    a1 = r.a1,
                    a2 = r.a2,
                    a3 = r.a3,
                    a4 = r.a4,
                    label = r.label,
                    binomial_label=r.binomial_label,
                    features = Vectors.dense(r.a1, r.a2, r.a3, r.a4))
                    ).toDF()


kmeans_estimator = KMeans()\
    .setFeaturesCol("features")\
    .setPredictionCol("prediction")\
kmeans_transformer = kmeans_estimator.fit(df)

predicted_df = kmeans_transformer.transform(df).drop("features") …

我在PySpark(ML包)中训练了LogisticRegression模型,预测结果是PySpark DataFrame(cv_predictions)(参见[1]).该probability列(见[2])是一种vector类型(见[3]).

[1]
type(cv_predictions_prod)
pyspark.sql.dataframe.DataFrame

[2]
cv_predictions_prod.select('probability').show(10, False)
+----------------------------------------+
|probability                             |
+----------------------------------------+
|[0.31559134817066054,0.6844086518293395]|
|[0.8937864350711228,0.10621356492887715]|
|[0.8615878905395029,0.1384121094604972] |
|[0.9594427633777901,0.04055723662220989]|
|[0.5391547673698157,0.46084523263018434]|
|[0.2820729747752462,0.7179270252247538] |
|[0.7730465873083118,0.22695341269168817]|
|[0.6346585276598942,0.3653414723401058] |
|[0.6346585276598942,0.3653414723401058] |
|[0.637279255218404,0.362720744781596]   |
+----------------------------------------+
only showing top 10 rows

[3]
cv_predictions_prod.printSchema()
root
 ...
 |-- rawPrediction: vector (nullable = true)
 |-- probability: vector (nullable = true)
 |-- prediction: double (nullable = true)

如何创建解析vectorPySpark DataFrame,以便创建一个新列,只拉取每个probability向量的第一个元素？

这个问题类似于,但下面链接中的解决方案不起作用/我不清楚:

如何在PySpark中访问denseVector的值

如何访问Spark DataFrame中VectorUDT列的元素？

我正在尝试在PySpark中运行线性回归,我想创建一个包含汇总统计信息的表,例如我的数据集中每列的系数,P值和t值.但是,为了训练线性回归模型,我必须使用Spark创建一个特征向量VectorAssembler,现在对于每一行我都有一个特征向量和目标列.当我尝试访问Spark的内置回归摘要统计信息时,它们会为每个统计信息提供一个非常原始的数字列表,并且无法知道哪个属性对应于哪个值,这很难通过手动计算出来大量的列.如何将这些值映射回列名？

例如,我的当前输出是这样的:

系数:[ - 187.807832407,-187.058926726,85.1716641376,10595.3352802,-127.258892837,-39.2827730493,-1206.47228704,33.7078197705,99.9956812528]

P值:[0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.18589731365614548,0.275173571416679,0.0]

t统计量:[ - 23.348593508995318,-44.72813283953004,19.836508234714472,144.49248881747755,-16.547272230754242,-9.560681351483941,-19.563547400189073,1.3232383890822680,1.0912415361190977,20.383256127350474]

系数标准误差:[8.043646497811427,4.182131353367049,4.293682291754585,73.32793120907755,7.690626652102948,4.108783841348964,61.669402913526625,25.481445101737247,91.63478289909655,609.7007361468519]

除非我知道它们对应哪个属性,否则这些数字毫无意义.但在我看来,DataFrame我只有一个名为"features"的列,其中包含稀疏向量行.

当我有一个热编码特征时,这是一个更大的问题,因为如果我有一个长度为n的编码变量,我会得到n个相应的系数/ p值/ t值等.

使用Spark ML变压器,我到达了 DataFrame每一行位置:

Row(object_id, text_features_vector, color_features, type_features)

其中text_features是一个稀疏的术语权重向量,color_features是一个小的20元素(一个热编码器)密集的颜色矢量,和type_features也是一种热编码器密集的矢量类型.

一个好的方法(使用Spark的设施)将这些特征合并到一个单独的大型数组中,以便测量任何两个对象之间的余弦距离？

我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 …