python - 大量列的性能下降。派斯帕克

Question

我在处理 spark 宽数据帧(大约 9000 列，有时更多)时遇到了问题。
任务:

1. 通过 groupBy 和 pivot 创建宽 DF。
2. 将列转换为向量并处理为来自 pyspark.ml 的 KMeans。

所以我制作了广泛的框架并尝试使用 VectorAssembler 创建矢量，缓存它并在其上训练 KMeans。
在独立模式下，我的电脑上的 7 个不同簇数的组装大约需要 11 分钟，KMeans 大约需要 2 分钟，帧为 ~500x9000。另一方面，pandas 中的这种处理(pivot df，并迭代 7 个集群)花费的时间不到一分钟。
显然我理解独立模式和缓存等的开销和性能下降，但这真的让我气馁。
有人可以解释一下如何避免这种开销吗？
人们如何使用 wide DF 而不是使用 vectorassembler 并导致性能下降？
更正式的问题(对于 sof 规则)听起来像 - 我怎样才能加快这段代码的速度？

%%time
tmp = (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 s

print(tmp.count(), len(tmp.columns))
552, 9378

%%time
transformed = assembler.transform(tmp).select('User', 'features').cache()
Wall time: 10min 45s

%%time
lst_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") \

Answer 1

VectorAssembler's transform function除了原始数据之外，还处理所有列并在每个列上存储元数据。这需要时间，还会占用 RAM。

要准确了解增加了多少，您可以将转换前后的数据帧转储为 parquet 文件并进行比较。根据我的经验，与 VectorAssembler 构建的特征向量相比，手动构建的特征向量或其他特征提取方法可能导致大小增加 10 倍，而这是针对只有 10 个参数的逻辑回归。如果数据集的列数与您的列数一样多，情况会变得更糟。

一些建议:

• 看看您是否可以通过其他方式构建特征向量。我不确定这在 Python 中的性能如何，但我已经从 Scala 中的这种方法中学到了很多东西。我注意到，对于手动构建的向量或使用其他提取方法 (TF-IDF) 而非 VectorAssembled 方法构建的向量，与逻辑回归(10 个参数)相比，性能差异大约为 5 到 6 倍。
• 看看您是否可以 reshape 数据以减少需要由 VectorAssembler 处理的列数。
• 看看增加 Spark 可用的 RAM 是否有帮助。