Spark ML Kmeans give : org.apache.spark.SparkException:无法执行用户定义的函数($anonfun$2：(vector) => int)
Spark ML Kmeans是Spark机器学习库中的一种聚类算法，用于将数据集划分为不同的簇。在使用Spark ML Kmeans时，有时可能会遇到org.apache.spark.SparkException:无法执行用户定义的函数($anonfun$2：(vector) => int)的错误。
这个错误通常是由于用户定义的函数无法正确执行导致的。用户定义的函数是指在使用Kmeans算法时，为了对数据进行处理或者自定义聚类逻辑而编写的函数。
要解决这个问题，可以按照以下步骤进行：
- 检查用户定义的函数：首先，需要仔细检查用户定义的函数($anonfun$2：(vector) => int)的实现是否正确。确保函数的输入和输出类型与算法要求的一致，并且函数逻辑正确。
- 检查输入数据：确保输入数据的格式和类型与算法要求的一致。Spark ML Kmeans要求输入数据为向量类型，如果输入数据不是向量类型，可能会导致无法执行用户定义的函数的错误。
- 检查Spark版本和依赖：确保使用的Spark版本和相关依赖库的版本与Spark ML Kmeans兼容。不同版本的Spark可能对函数的执行有不同的要求，如果版本不兼容可能会导致无法执行用户定义的函数的错误。
- 查看Spark日志：如果以上步骤都没有解决问题，可以查看Spark的日志文件，查找更详细的错误信息。Spark的日志文件通常位于日志目录下的spark目录中，可以根据具体的配置进行查找。
总结起来，当遇到org.apache.spark.SparkException:无法执行用户定义的函数($anonfun$2：(vector) => int)的错误时，需要检查用户定义的函数、输入数据、Spark版本和依赖以及查看Spark日志，以找到并解决问题。
关于Spark ML Kmeans的更多信息，您可以参考腾讯云的相关产品Spark MLlib：https://cloud.tencent.com/document/product/851/18385
相关·内容
1回答
我尝试加载KmeansModel，然后从中获取标签：
下面是我写的代码：
val kMeansModel = KMeansModel.load(trainedMlModel.mlModelFilePath)
val arrayOfElements = measurePoint.measurements.map(a => a._2).toSeq
println(s"ArrayOfELements::::$arrayOfElements")
val arrayDF = sparkContex
浏览 6提问于2017-03-09得票数 1
回答已采纳
我是scala和mllib的新手，我收到了以下错误。如果有人能解决类似的问题，请告诉我。
import org.apache.spark.sql.SparkSession
import org.apache.spark.mllib.clustering.{KMeans, KMeansModel}
import org.apache.spark.mllib.linalg.Vectors
.
.
.
val conf = new SparkConf().setMaster("local").setAppName("SampleApp")
val sContext =
浏览 0提问于2018-11-28得票数 2
1回答
虽然运行K意味着使用pyspark进行集群，但我使用以下代码行来查找最佳K值。但是在模型拟合线上经常会出现一些错误。
预处理阶段包括去除NAs和标记编码，
from pyspark.sql.functions import when,col
#Encode column "Potential" from dataframe df
high = list(range(86,101))
middle = list(range (71, 86))
low = list(range(56, 71))
very_low = list(range(45,56))
#checking
浏览 47提问于2020-06-01得票数 1
回答已采纳
下面有一个带有代码的笔记本，它会引发以下错误：
找不到参数sparkSession的隐式值
import org.apache.spark.sql.{SparkSession, Row, DataFrame}
import org.apache.spark.ml.clustering.KMeans
def createBalancedDataframe(df:DataFrame, reductionCount:Int)(implicit sparkSession:SparkSession) = {
val kMeans = new KMeans().setK(reducti
浏览 3提问于2019-03-15得票数 0
当我运行以下代码行时
>model_pca <- tbl(sc, "flights") %>% select(air_time,distance,dep_time) %>%
ml_pca()
结果导致阶段失败，但我不能理解原因
org.apache.spark.SparkException: Failed to execute user defined function($anonfun$3: (struct<air_time:double,distance:double,dep_time_double_vector_assembler_
浏览 0提问于2018-03-08得票数 0
1回答
我是从存储在HDFS上的文件夹中读取的流数据。我有以下一小部分代码：
// Convert text into a DataSet of LogEntry rows. Select the two columns we care about
val df = rawData.flatMap(parseLog).select("ip", "status")
df .isStreaming
val kmeans = new KMeans().setK(2).setSeed(1L)
val model = kmeans.fit(df)
//
浏览 0提问于2018-03-09得票数 1
在执行prep.show()/pred.head(2)时出现错误。我有如下的df_train和df_test。pyspark随机森林分类器的数据帧格式是否错误？ >>> df_train.show(n=2)
+-----+--------------------+
|label| features|
+-----+--------------------+
| 1.0|[6.0,148.0,72.0,3...|
| 0.0|[1.0,85.0,66.0,29...|
+-----+--------------------+
only showin
浏览 42提问于2021-04-22得票数 0
1回答
我有一个例子，想在UDF中创建Dataframe。类似于下面的内容
import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.ml.feature.VectorAssembler
数据到数据帧
val df = Seq((1,1,34,23,34,56),(2,1,56,34,56,23),(3,0,34,23,23,78),(4,0,23,34,78,23),(5,1,56,23,2
浏览 2提问于2018-11-30得票数 1
AWS EC2 Spark / Hadoop集群。
下面的baisc K-表示与Spark2.0.1一起工作的sparklyr代码
kmeans_model <- iris_tbl %>%
select(Petal_Width, Petal_Length) %>%
ml_kmeans(centers = 3)
我已经升级到Spark2.1.1，并得到了这个错误
Error: java.lang.IllegalArgumentException: Field "features" does not exist.
at org.
浏览 2提问于2017-11-27得票数 3
回答已采纳
1回答
有些人能帮我克服错误吗？
kmeans_model <- iris_tbl %>%
select(Petal_Width, Petal_Length) %>%
ml_kmeans(centers = 3)
错误: java.lang.IllegalArgumentException:字段“功能”不存在。可用字段: Petal_Width、Petal_Length
在org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:274) at org.apache.spa
浏览 0提问于2019-03-31得票数 0
回答已采纳
1回答
我正在运行kmeans算法，我创建了一个VectorAssembler，将inputcols设置为(“经度”，“纬度”)，并将outputCol设置为("location")。我需要将我的数据从一个json文件集群到3个集群。我按经度和纬度对数据进行分类，并创建连接两者的Vector Location。位置和纬度为DoubleType。我认为这是因为位置向量，我得到了下面的错误： 19/04/08 15:20:56 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
org.apache.spark.Spark
浏览 16提问于2019-04-08得票数 1
在对MultilayerPerceptronClassifier的layers[]参数使用不同的值时，我继续遇到一些奇怪的错误。
例如，对于相同的数据：
int[] layers = {100, 98, 2}
new MultilayerPerceptronClassifier().setLayers(layers).setLabelCol(targetColumn).fit(data);
我明白了: java.lang.ArrayIndexOutOfBoundsException
With stack trace:
at org.apache.spark.scheduler.D
浏览 0提问于2017-02-08得票数 2
我在一个数据集上运行逻辑回归，看起来一切正常，但当我要打印出混淆矩阵时，我得到了一个错误，我不确定如何处理。
import org.apache.spark.mllib.evaluation.MulticlassMetrics
val predictionAndLabels = results.select($"prediction", $"label").as[(Double, Double)].rdd
val metrics = new MulticlassMetrics(predictionAndLabels)
println("Confusi
浏览 0提问于2016-12-28得票数 2
大家好，祝您愉快。根据你的经验，我想得到一些帮助。我正在尝试将文本文档集合转换为基于CountVectorizerModel数组大小为24693的自定义词汇表的令牌计数向量。
这里是简单的代码
CountVectorizerModel cvm2 = new CountVectorizerModel(vocabulary)
.setInputCol(NEXT)
.setOutputCol(NEXT_RAW_FEATURES);
cvm2.transform(dataset).show(false);
这里是我的全部例
浏览 4提问于2018-01-10得票数 1
1回答
规范一列dataframe
、、、、
我试图使用L1库对数据框架中列的值进行pyspark ML规范化。以下是我的密码。但它做不到。你能帮我弄清楚这段代码有什么问题吗？
from pyspark.ml.feature import Normalizer
y = range(1,10)
data = spark.createDataFrame([[float(e), ] for e in y])
#data.select('_1').show()
normalizer = Normalizer(p=1.0, inputCol="_1", outputCol="features")
浏览 0提问于2017-06-09得票数 1
回答已采纳
Exception in thread "main" java.lang.IllegalArgumentException: Field "features" does not exist.
at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:264)
at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:264)
at s
浏览 2提问于2017-06-13得票数 2
我是Spark和MLlib的新手，我试图从我的java应用程序中调用StreamingKMeans，但我得到了一个我似乎不理解的异常。下面是我用来转换训练数据的代码：
JavaDStream<Vector> trainingData = sjsc.textFileStream("/training")
.map(new Function<String, Vector>() {
public DenseVector call(String line) throws Exception {
浏览 1提问于2015-06-10得票数 6
我正努力从我的火花源实现机器学习(kmeans)。我有一个表，有2列:回顾和标签(正面或负面)，一切似乎都很好。但是，当我运行预测时，我会得到下一个错误：
SparkException:由于阶段失败而中止的任务:阶段22.0中的任务0失败了1次，最近的失败:在阶段22.0中丢失了任务0.0 (TID 22，localhost)：org.apache.spark.SparkException:未见标签
以下是代码：
sc <- spark_connect(master = "local", version="2.0.0")
colnames(dfA
浏览 0提问于2019-01-25得票数 1
1回答
我试图在scala中定义一个函数，以便使用Spark在其上进行迭代。这是我的代码：
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.m
浏览 4提问于2016-05-13得票数 0
回答已采纳
当我试图从Vector Transformer的输出创建标签点时，我面临以下问题：
val realout = output.select("label","features").rdd.map(row => LabeledPoint
row.getAs[Double]("label"),
row.getAs[org.apache.spark.mllib.linalg.SparseVector]("features")
))
我得到的错误是：
enter [error] (run-main-0) org.apach
浏览 0提问于2016-11-12得票数 4
我使用下面的代码来规范PySpark DataFrame
from pyspark.ml.feature import StandardScaler, VectorAssembler
from pyspark.ml import Pipeline
cols = ["a", "b", "c"]
df = spark.createDataFrame([(1, 0, 3), (2, 3, 2), (1, 3, 1), (3, 0, 3)], cols)
Pipeline(stages=[
VectorAssembler(inputCols=
浏览 5提问于2016-12-21得票数 3
回答已采纳
我正在尝试从(大型)文本文档集合()上运行一个KMeans在MLLib上。文档通过Lucene分析器发送，稀疏向量由HashingTF.transform()函数创建。无论我使用的并行化程度如何(通过合并函数)，KMeans.train总是在下面返回一个OutOfMemory异常。对如何解决这个问题有什么想法吗？
Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
at scala.reflect.ManifestFactory$$anon$12.newArray(Manifest.sca
浏览 0提问于2014-10-19得票数 4
1回答
Kmeans计算成本
、、、、
我正在使用这个，它不是我写的。为了预测质心，我不得不这样做：
model = cPickle.load(open("/tmp/model_centroids_128d_pkl.lopq"))
codes = d.map(lambda x: (x[0], model.predict_coarse(x[1])))
其中`d.first()‘产生如下结果：
(u'3768915289',
array([ -86.00641097, -100.41325623, <128 coords in total>]))
和codes.first()
(u
浏览 13提问于2016-08-29得票数 4
回答已采纳
我正在研究一个使用Spark上不同分类器的二进制分类问题；能够成功地训练和评估模型(如朴素贝叶斯、随机森林、Logistic Reg等)，然而，我在使用相同的培训和测试数据开发多层Perceptron分类器时遇到了一些问题。
也许能帮我找出我哪里出了问题！
# spark version
sc.version
>>u'2.3.0.2.6.5.25-1'
# python version
import sys
print (sys.version)
>>2.7.5 (default, May 3 2017, 07:55:04)
[GCC 4.8.5
浏览 1提问于2018-09-29得票数 0
我在跟踪，当我执行以下行时，会得到这个错误
best_lr = lr.fit(training)
误差
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-102-88042cb88c20> in <module>()
----> 1 best_lr = lr.fit(training)
浏览 0提问于2018-12-04得票数 0
我有下面的模型，我想用SparkML MultilayerPerceptronClassifier()来估计。
val formula = new RFormula()
.setFormula("vtplus15predict~ vhisttplus15 + vhistt + vt + vtminus15 + Time + Length + Day")
.setFeaturesCol("features")
.setLabelCol("label")
formula.fit(data).transform(data)
注：特征是一个
浏览 4提问于2017-12-15得票数 1
回答已采纳
..。
val cols: Seq[String] = Seq("item", "SR", "RP")
val vecToSeq = udf((v:org.apache.spark.ml.linalg.Vector) => v.toArray)
val exprs = cols.zipWithIndex.map{ case(c,i) => $"_tmp".getItem(i).alias(c)}
val DoubleDF = result5.select(vecToSeq($"vectorCol").
浏览 0提问于2017-02-28得票数 0
1回答
我有一些关于服务提供商客户的数据(~1MB)。我正在尝试根据几个特性来预测是否会终止订阅(PySpark on Databricks)。
单特征模型
首先，我只试了一个功能，并看到了成功的培训：
# Create vector assembler to merge independent features (in this case just one) into one feature as a list
vectorAssembler = VectorAssembler(inputCols=['MonthlyCharges'], outputCol='Charges&
浏览 2提问于2021-07-23得票数 3
回答已采纳
我试着用kmeans做一个聚类操作。我的数据集是：
我没有多少火花方面的经验，我只工作了几个月，当我尝试应用kmean.train时会发生错误，它有一个输入:向量、num_cluster和迭代。
我正在本地运行，我的机器有可能不能计算这么多数据吗？
主要代码是：
import org.apache.spark.sql.SparkSession
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.collection._
import org.apache.spark.sql.fu
浏览 0提问于2018-05-20得票数 1
1回答
我已经创建了一条流水线，并试图在spark中训练Kmean聚类算法，但是它失败了，我无法找到确切的错误是什么。这是代码
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.ml.evaluation.ClusteringEvaluator
import org.apache.spark.ml.feature.{OneHotEncoderEstimator, StringIndexer, VectorAssembler, Normalize
浏览 1提问于2020-04-16得票数 1
回答已采纳
2回答
我有下面的表，为了执行联接，我正在为它生成一个带有rowId列的数字序列，但这会引发以下错误。我做错了什么？请帮我处理这个。
fListVec: org.apache.spark.sql.DataFrame = [features: vector]
+-----------------------------------------------------------------------------+
|features |
+--------------
浏览 7提问于2017-11-30得票数 1
回答已采纳
我正在尝试对两个数据帧进行字符串匹配。假设dataframe1包含X个句子和dataframe2 Y个句子。我需要检查一下，Dataframe1中的任何句子都与Dataframe2匹配。我正在尝试使用ML管道，如下所示：
def match_names(df_1, df_2):
pipeline = Pipeline(stages=[
RegexTokenizer(
pattern="", inputCol="name", outputCol="tokens", minTokenLength=1
浏览 2提问于2019-04-11得票数 1
1回答
我正在尝试将向量数据类型提供给一个名为mllib的Word2Vec函数。当Word2Vec返回一个包含所需向量的“结果”列的DataFrame时，需要一些代码。最后，当代码在Spark中成功运行时，我尝试使用.foreach来println几行代码。火花在这一步崩溃，有以下错误：NullPointerException。如果删除println命令，代码将运行良好。我试过使用RDD的示例方法，但同样的火花错误出现了。不知何故，RDD变得不可读了。
若要了解此ML任务的背景，请参阅此。
import org.apache.spark._
import org.apache.spark.rdd._
浏览 0提问于2018-03-19得票数 1
回答已采纳
我正在尝试使用匿名函数将JavaRDD<String>转换为JavaRDD<Row>。下面是我的代码：
JavaRDD<String> listData = jsc.textFile("/src/main/resources/CorrectLabels.csv");
JavaRDD<Row> jrdd = listData.map(new Function<String, Row>() {
public Row call(String record) throws Excepti
浏览 0提问于2016-05-25得票数 1
我遵循了- 中的一个示例
当我试图获取模型摘要时，我遇到了一个错误。这是我的错误代码-
// START
import org.apache.spark.ml.classification.LogisticRegression
// Load training data
val training = spark.read.format("libsvm").load("file:///Users/my_username/Desktop/sample_multiclass_classification_data.txt")
val lr = new Logi
浏览 0提问于2018-12-27得票数 4
我有跟踪df -
+--------+----------+
| Id|prediction|
+--------+----------+
| 1| 0.0|
| 2| 0.0|
| 3| 1.0|
| 4| 1.0|
| 5| 0.0|
| 6| 0.0|
| 7| 0.0|
| 8| 0.0|
| 9| 0.0|
| 10| 0.0|
+----
浏览 1提问于2018-01-31得票数 0
回答已采纳
我在KMeans.train()内部遇到一个异常，如下所示：
java.lang.IllegalArgumentException: requirement failed
at scala.Predef$.require(Predef.scala:212)
at org.apache.spark.mllib.util.MLUtils$.fastSquaredDistance(MLUtils.scala:487)
at org.apache.spark.mllib.clustering.KMeans$.fastSquaredDistance(KMeans.scala:589)
a
浏览 0提问于2017-10-27得票数 4
在Windows上火花2.1 (独立)。添加VectorAssembler列后，无法将spark保存到parquet文件。在向量列之前保存数据没有问题，所有“功能”都不是空的(使用NVL)
val conf = new SparkConf().setAppName("RandomForestModelML").setMaster("local")
val sparkSession = SparkSession.builder().config(conf).getOrCreate()
val df = sparkSession.read
.o
浏览 2提问于2017-02-25得票数 1
我正在尝试为Pyspark中的Kmeans模型创建并加载pickle文件。我使用的是Python3.7.9和PySpark版本3.0.1。我可以创建一个pickle文件，但出现以下错误： 代码： from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.clustering import KMeans
import joblib
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('Sess
浏览 62提问于2021-01-12得票数 0
1回答
在某些文本挖掘之后，我试图进行一次k均值聚类，但是我找不到如何在ParseWikipedia.termDocumentMatrix方法所需的数据集中转换kmean.fit的结果
scala> val (termDocMatrix, termIds, docIds, idfs) = ParseWikipedia.termDocumentMatrix(lemmas, stopWords, numTerms, sc)
scala> val kmeans = new KMeans().setK(5).setMaxIter(200).setSeed(1L)
scala> termDoc
浏览 3提问于2018-01-05得票数 1
回答已采纳
我想通过设置epsilon=1e-4而不是设置numIterations来训练火花上的K均值模型。在火花壳中，我输入：
val model = KMeans.train(trainRDD, numClusters=8, runs=30, initializationMode="k-means||",epsilon=1e-4)
但是，错误信息如下所示：
scala> val model = KMeans.train(trainRDD, numClusters=8, runs=30, initializationMode="k-means||",epsilon
浏览 0提问于2016-01-11得票数 0
回答已采纳
2回答
我正在为我的项目使用cassandra和spark，现在我编写了这个来从DB检索数据：
results = session.execute("SELECT * FROM foo.test");
ArrayList<String> supportList = new ArrayList<String>();
for (Row row : results) {
supportList.add(row.getString("firstColumn") + "," + row.getString(&
浏览 5提问于2015-07-30得票数 0
回答已采纳
2回答
当我把卡夫卡和火花streaming.When集成在一起时，我遇到了广播变量的问题，我没有使用火花广播，卡夫卡和火花流集成没有问题，然后我使用广播，它是错误的。我使用星火集群作为独立的。我的scala代码是这样的
object test {
//spark streaming paramters
var conf = new SparkConf().setAppName("SPK_test").setMaster("spark://192.168.90.21:7077")
var sc = new SparkContext(conf)
var
浏览 1提问于2018-09-10得票数 0
回答已采纳
在我们的一个Pyspark作业中，我们有一个场景，我们在一个大的数据帧和相对较小的数据帧之间进行连接，我相信spark正在使用广播连接，我们遇到了以下错误 org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 8 GB
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec$$anonfun$relationFuture$1$$anonfun$apply$1.apply(BroadcastExch
浏览 752提问于2020-04-22得票数 0
我尝试使用结构内部的Vector作为spark mllib转换的输入列。像这样..。
import org.apache.spark.ml.linalg._
case class State(id: String, features: Vector)
val ds = Seq[(State,State)]().toDS
ds.printSchema()
root
|-- _1: struct (nullable = true)
| |-- id: string (nullable = true)
| |-- features: vector (nullable = true)
|
浏览 21提问于2019-07-16得票数 0
我想使用StandardScaler来标准化这些特性。
下面是我的代码：
val Array(trainingData, testData) = dataset.randomSplit(Array(0.7,0.3))
val vectorAssembler = new VectorAssembler().setInputCols(inputCols).setOutputCol("features").transform(trainingData)
val stdscaler = new StandardScaler().setInputCol("features&
浏览 2提问于2016-12-28得票数 15
回答已采纳
我正在尝试并行运行大量的k-means。我有一个房间和它的大量数据，我想计算每个房间的集群。所以我有
roomsSignals[(room:String, signals:List[org.apache.spark.mllib.linalg.Vector]]
roomsSignals.map{l=>
val data=sc.parallelize(l.signals)
val clusterCenters=2
val model = KMeans.train(data, clusterCenters, 5)
model.clusterCenters.map { r =>
浏览 0提问于2016-12-13得票数 1
我正在尝试在多个列上应用字符串索引器。以下是我的代码
val stringIndexers = Categorical_Model.map { colName =>new StringIndexer().setInputCol(colName).setOutputCol(colName + "_indexed")}
var dfStringIndexed = stringIndexers(0).fit(df3).transform(df3) // 'fit's a model then 'transform's data
for(x<
浏览 77提问于2019-07-22得票数 7
我试图使用火花数据作为输入我的k-均值模型。不管怎么说，我总是犯错误。(检查代码后的部分)
我的看起来像这样(大约有100万行)：
ID col1 col2 Latitude Longitude
13 ... ... 22.2 13.5
62 ... ... 21.4 13.8
24 ... ...
浏览 2提问于2017-07-06得票数 2
我有一个名为pres2的org.apache.spark.sql.DataFrame (大约1000行，在运行Spark 2.1.0的AWS3节点EMR集群(v5.3.1)上生成)，如下所示：
+-------------------+--------------------+----------+
| CreatedAt| tweet|prediction|
+-------------------+--------------------+----------+
|2015-10-12 20:58:52|This Saturday is ...
浏览 2提问于2017-03-11得票数 9
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
