python之Pyspark 中的宽数据帧操作太慢

我是 Spark 新手，正在尝试使用 pyspark (Spark 2.2) 对非常广泛的功能集(约 1300 万行，15000 列)执行过滤和聚合操作。功能集作为 parquet 文件存储在 S3 驱动器上。我正在运行一个测试脚本来加载数据框中的特征集，选择几千条记录，按特定区域代码分组，并对 15k 特征列中的每一个进行平均。问题是作业出错或花费的时间太长(对于 5% 的记录样本大约需要 8 小时)。

有没有什么方法可以在 Pyspark 的宽数据帧上加速这些类型的操作？我正在使用 Jupyter notebooks 并希望这些查询在几分钟而不是几小时内完成。

这是我的代码

df_feature_store = spark.read.parquet(PATH_FEATURE_STORE).sample(False, 0.05, seed=0).cache() 
    logger.info("Initial data set loaded and sampled") 
 
    df_selected_rors = spark.read.csv(PATH_DATA_SOURCE+"ROR Sample.csv", header=True) 
    agg_cols = [x for x in df_feature_store.columns if re.search("^G\d{2}",x)] 
    agg_cols = agg_cols[:10]  # just testing with fewer columns 
    expr = {x:"mean" for x in agg_cols} 
    joineddf = df_feature_store.join(df_selected_rors, df_feature_store.ROLLOUTREGION_IDENTIFIER == df_selected_rors.ROR, "inner") 
    aggdf = joineddf.groupby("ROLLOUT_REGION_IDENTIFIER").agg(expr) 
    # replace groupby 
    # loop for a 1000 column aggregations  
    # transpose columns into rows as arrays 
    aggdf.write.mode("overwrite").csv(PATH_FEATURE_STORE + "aggregated", header=True) 
    logger.info("Done")` 

请您参考如下方法:

我会尝试将其拆分以查看问题所在

• 某些版本的 Spark 与 DF 中的许多列存在问题；我不记得具体情况了。
• 在任何查询之前，从 CSV 读取并保存在本地 Parquet 中，如果可以的话过滤列
• 运行查询 Parquet local- 到 Parquet local

作为工作目的地的 S3 (a) 提交速度慢，并且 (b) 由于 S3 的最终一致性而面临丢失数据的风险。除非您使用 S3mper/S3Guard/EMR 一致的 EMRFS，否则您不应将其用作工作的直接目标。