【发布时间】：2018-01-27 05:35:32
【问题描述】：
我的程序流程是这样的：
1. 将 parquet 文件中的 40 亿行 (~700GB) 数据读取到数据框中。使用的分区大小为 2296
2. 清理并过滤掉 25 亿行
3. 使用管道模型和经过训练的模型转换剩余的 15 亿行。该模型使用逻辑回归模型进行训练，其中预测 0 或 1，并且 30% 的数据从转换后的数据框中过滤掉。
4. 上述数据框与另一个约 1 TB 的数据集左外连接（也从 parquet 文件中读取。）分区大小为 4000
5. 加入另一个大约 100 MB 的数据集，例如 joined_data = data1.join(broadcast(small_dataset_100MB), data1.field == small_dataset_100MB.field, "left_outer")
6. 然后将上述数据框分解为 ~2000 的因子 exploded_data = joined_data.withColumn('field', explode('field_list'))
7. 进行聚合 aggregate = exploded_data.groupBy(*cols_to_select)\
.agg(F.countDistinct(exploded_data.field1).alias('distincts'), F.count("*").alias('count_all')) cols_to_select 列表共有 10 列。
8. 最后执行一个动作，aggregate.count()。
问题是，倒数第三阶段（200 个任务）永远卡在任务 199。尽管分配了 4 个核心和 56 个执行器，但计数仅使用一个核心和一个执行器来运行作业。我尝试将大小从 40 亿行分解为 7 亿行，这是 1/6 部分，花了四个小时。对于如何加快此过程，我非常感谢您的帮助。谢谢
【问题讨论】：
标签： apache-spark pyspark spark-dataframe hadoop-yarn