Spark的StringIndexer非常有用,但是通常需要检索生成的索引值和原始字符串之间的对应关系,看起来应该有一种内置的方法来完成它.我将使用Spark文档中的这个简单示例来说明:
from pyspark.ml.feature import StringIndexer
df = sqlContext.createDataFrame(
[(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")],
["id", "category"])
indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
indexed_df = indexer.fit(df).transform(df)
这个简化的案例给了我们:
+---+--------+-------------+
| id|category|categoryIndex|
+---+--------+-------------+
| 0| a| 0.0|
| 1| b| 2.0|
| 2| c| 1.0|
| 3| a| 0.0|
| 4| a| 0.0|
| 5| c| 1.0|
+---+--------+-------------+
所有精细和花花公子,但对于许多用例,我想知道我的原始字符串和索引标签之间的映射.我能想到的最简单的方法就是这样:
In [8]: indexed.select('category','categoryIndex').distinct().show()
+--------+-------------+
|category|categoryIndex|
+--------+-------------+
| b| 2.0|
| c| 1.0|
| a| 0.0|
+--------+-------------+ …python apache-spark apache-spark-sql pyspark apache-spark-ml
最近我打算将我的独立python ML代码迁移到spark.ML管道spark.ml非常方便,使用简化的API来链接算法阶段和超参数网格搜索.
尽管如此,我发现它支持现有文档中的一个重要特征:缓存中间结果.当管道涉及计算密集阶段时,会出现此功能的重要性.
例如,在我的情况下,我使用一个巨大的稀疏矩阵来对时间序列数据执行多个移动平均,以便形成输入要素.矩阵的结构由一些超参数确定.这一步结果是整个管道的瓶颈,因为我必须在运行时构造矩阵.
在参数搜索期间,我通常还有其他参数来检查除"结构参数"之外的其他参数.因此,如果我可以在"结构参数"不变的情况下重复使用巨大的矩阵,那么我可以节省大量的时间.出于这个原因,我故意形成了我的代码来缓存和重用这些中间结果.
所以我的问题是:Spark的ML管道可以自动处理中间缓存吗？或者我是否必须手动编写代码才能这样做？如果是这样,有什么最佳实践可供学习吗？
PS我已经查看了官方文档和其他一些材料,但似乎没有人讨论这个主题.
在我训练了LogisticRegressionModel之后,我用它转换了测试数据DF并得到了预测DF.然后当我调用prediction.show()时,输出列名称为:[label | features | rawPrediction | probability | prediction].我知道label和featrues意思,但我应该如何理解rawPrediction|probability|prediction？
我试图用Zeppelin在Spark ML中建立一个模型.我是这个领域的新手,想要一些帮助.我想我需要将正确的数据类型设置为列并将第一列设置为标签.非常感谢任何帮助,谢谢
val training = sc.textFile("hdfs:///ford/fordTrain.csv")
val header = training.first
val inferSchema = true
val df = training.toDF
val lr = new LogisticRegression()
.setMaxIter(10)
.setRegParam(0.3)
.setElasticNetParam(0.8)
val lrModel = lr.fit(df)
// Print the coefficients and intercept for multinomial logistic regression
println(s"Coefficients: \n${lrModel.coefficientMatrix}")
println(s"Intercepts: ${lrModel.interceptVector}")
我正在使用的csv文件的片段是:
IsAlert,P1,P2,P3,P4,P5,P6,P7,P8,E1,E2
0,34.7406,9.84593,1400,42.8571,0.290601,572,104.895,0,0,0,
有没有办法从pyspark中的Spark ML获取ROC曲线上的点？在文档中，我看到了一个 Scala 的例子，但不是 python：https : //spark.apache.org/docs/2.1.0/mllib-evaluation-metrics.html
那正确吗？我当然可以想出实现它的方法，但我不得不想象如果有一个预先构建的函数它会更快。我正在处理 300 万个分数和几十个模型，所以速度很重要。
.map()我用python中的哪个函数labeledPoints从spark数据帧创建一组？如果标签/结果不是第一列,但我可以参考其列名"状态",那么表示法是什么？
我用这个.map()函数创建了Python数据帧:
def parsePoint(line):
listmp = list(line.split('\t'))
dataframe = pd.DataFrame(pd.get_dummies(listmp[1:]).sum()).transpose()
dataframe.insert(0, 'status', dataframe['accepted'])
if 'NULL' in dataframe.columns:
dataframe = dataframe.drop('NULL', axis=1)
if '' in dataframe.columns:
dataframe = dataframe.drop('', axis=1)
if 'rejected' in dataframe.columns:
dataframe = dataframe.drop('rejected', axis=1)
if 'accepted' in dataframe.columns:
dataframe = dataframe.drop('accepted', axis=1)
return dataframe
在reduce函数重新组合了所有Pandas数据帧之后,我将其转换为Spark数据帧.
parsedData=sqlContext.createDataFrame(parsedData)
但是现在我如何labledPoints用Python 创建呢？我想它可能是另一个.map()功能？
python pandas apache-spark apache-spark-ml apache-spark-mllib
我正在尝试调整使用隐式数据的ALS矩阵分解模型的参数.为此,我正在尝试使用pyspark.ml.tuning.CrossValidator来运行参数网格并选择最佳模型.我相信我的问题在于评估者,但我无法弄明白.
我可以使用回归RMSE评估器为显式数据模型工作,如下所示:
from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.sql.functions import rand
conf = SparkConf() \
.setAppName("MovieLensALS") \
.set("spark.executor.memory", "2g")
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
dfRatings = sqlContext.createDataFrame([(0, 0, 4.0), (0, 1, 2.0), (1, 1, 3.0), (1, 2, 4.0), (2, 1, 1.0), (2, 2, 5.0)],
["user", "item", "rating"])
dfRatingsTest = sqlContext.createDataFrame([(0, 0), (0, 1), (1, 1), (1, 2), (2, …尝试构建ML时出现以下错误Pipeline:
pyspark.sql.utils.IllegalArgumentException: 'requirement failed: Column features must be of type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 but was actually ArrayType(DoubleType,true).'
我的features列包含一个浮点值数组.听起来我需要将它们转换为某种类型的向量(它不是稀疏的,所以是DenseVector？).有没有办法直接在DataFrame上执行此操作,还是需要转换为RDD？
python apache-spark pyspark apache-spark-ml apache-spark-mllib
您好，我正在与 Pyspark 合作，第一次使用 ML 包实施情感分析项目，cofde 工作正常，但突然显示上面提到的错误有人可以帮忙吗这是完整的错误描述
ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:50532)
Traceback (most recent call last):
File "C:\opt\spark\spark-2.3.0-bin-hadoop2.7\python\lib\py4j-0.10.6-src.zip\py4j\java_gateway.py", line 852, in _get_connection
connection = self.deque.pop()
IndexError: pop from an empty deque
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\opt\spark\spark-2.3.0-bin-hadoop2.7\python\lib\py4j-0.10.6-src.zip\py4j\java_gateway.py", line 990, in start
self.socket.connect((self.address, self.port))
ConnectionRefusedError: [WinError 10061] Aucune connexion n’a pu être établie car l’ordinateur cible l’a expressément refusée
我是火花新手。我正在使用以下配置集在 Spark 独立版 (v3.0.0) 中编写机器学习算法：
SparkConf conf = new SparkConf();
conf.setMaster("local[*]");
conf.set("spark.driver.memory", "8g");
conf.set("spark.driver.maxResultSize", "8g");
conf.set("spark.memory.fraction", "0.6");
conf.set("spark.memory.storageFraction", "0.5");
conf.set("spark.sql.shuffle.partitions", "5");
conf.set("spark.memory.offHeap.enabled", "false");
conf.set("spark.reducer.maxSizeInFlight", "96m");
conf.set("spark.shuffle.file.buffer", "256k");
conf.set("spark.sql.debug.maxToStringFields", "100");
这就是我创建 CrossValidator 的方式
ParamMap[] paramGrid = new ParamGridBuilder()
.addGrid(gbt.maxBins(), new int[]{50})
.addGrid(gbt.maxDepth(), new int[]{2, 5, 10})
.addGrid(gbt.maxIter(), new int[]{5, 20, 40})
.addGrid(gbt.minInfoGain(), new double[]{0.0d, .1d, .5d})
.build();
CrossValidator gbcv = new CrossValidator()
.setEstimator(gbt)
.setEstimatorParamMaps(paramGrid)
.setEvaluator(gbevaluator)
.setNumFolds(5)
.setParallelism(8)
.setSeed(session.getArguments().getTrainingRandom());
问题是，当（在 paramGrid 中） maxDepth 只是 {2, 5} 和 maxIter {5, 20} 时，一切都工作得很好，但是当它像上面的代码中那样时，它会不断记录： ，其中 …