我在处理宽数据框时遇到了问题（大约9000列，有时更多）。
任务：
- 通过groupBy和pivot创建宽数据框。
- 将列转换为向量，并使用pyspark.ml中的KMeans进行处理。
因此，我创建了一个大数据框，并尝试使用VectorAssembler创建向量，对其进行缓存，并在其上训练KMeans。
在我的PC上以独立模式处理大约500行x9000列的数据框时，组装过程大约需要11分钟，KMeans需要2分钟，处理7个不同数量的聚类。
相比之下，使用pandas处理（pivot数据框，并迭代7个聚类）不到一分钟。
显然我理解独立模式、缓存等带来的开销和性能下降，但这真的让我很沮丧。
有人能解释一下我如何避免这种开销吗？
人们如何处理宽数据框而不使用VectorAssembler并避免性能下降？
更正式的问题（符合stackoverflow规则）听起来像 – 我如何加速这段代码？
%%timetmp = (df_states.select('ObjectPath', 'User', 'PropertyFlagValue') .groupBy('User') .pivot('ObjectPath') .agg({'PropertyFlagValue':'max'}) .fillna(0))ignore = ['User']assembler = VectorAssembler( inputCols=[x for x in tmp.columns if x not in ignore], outputCol='features')Wall time: 36.7 sprint(tmp.count(), len(tmp.columns))552, 9378%%timetransformed = assembler.transform(tmp).select('User', 'features').cache()Wall time: 10min 45s%%timelst_levels = []for num in range(3, 14): kmeans = KMeans(k=num, maxIter=50) model = kmeans.fit(transformed) lst_levels.append(model.computeCost(transformed))rs = [i-j for i,j in list(zip(lst_levels, lst_levels[1:]))]for i, j in zip(rs, rs[1:]): if i - j < j: print(rs.index(i)) kmeans = KMeans(k=rs.index(i) + 3, maxIter=50) model = kmeans.fit(transformed) break Wall time: 1min 32s配置：
.config("spark.sql.pivotMaxValues", "100000") \.config("spark.sql.autoBroadcastJoinThreshold", "-1") \.config("spark.sql.shuffle.partitions", "4") \.config("spark.sql.inMemoryColumnarStorage.batchSize", "1000") \回答：
实际上，解决方案是在map中找到的，用于RDD。
- 首先，我们要创建一个值的映射表。
- 同时提取所有不同的名称。
- 倒数第二步，我们在名称字典中搜索每行映射表的值，并返回值或如果未找到则返回0。
- 对结果应用向量组装器。
优点：
- 您不必创建具有大量列的数据框，从而避免了开销。（速度从11分钟提高到1分钟。）
- 您仍然在集群上工作，并在spark的范式中执行代码。
代码示例：scala实现。
