为什么 PySpark 任务花费了太多时间？答案

【问题标题】：Why PySpark task is taking too much time?为什么 PySpark 任务花费了太多时间？
【发布时间】：2020-07-17 15:36:36
【问题描述】：

我正在运行一个没有问题的 Pyspark 进程。该过程的第一步是将特定的 UDF 应用于数据帧。这是函数：

import html2text

class Udfs(object):
    def __init__(self):
        self.h2t = html2text.HTML2Text()
        self.h2t.ignore_links = True
        self.h2t.ignore_images = True

    def extract_text(self, raw_text):
        try:
            texto = self.h2t.handle(raw_text)
        except:
            texto = "PARSE HTML ERROR"
        return texto

这是我应用 UDF 的方式：

import pyspark.sql.functions as f
import pyspark.sql.types as t
from udfs import Udfs

udfs = Udfs()
extract_text_udf = f.udf(udfs.extract_text, t.StringType())
df = df.withColumn("texto", extract_text_udf("html_raw"))

它处理大约 2900 万行和 300GB。问题是某些任务需要太多时间来处理。任务的平均时间为：

其他任务已完成，持续时间超过 1 小时。

但有些任务需要太多时间处理：

该进程在 AWS 中使用 EMR 在具有 100 个节点的集群中运行，每个节点具有 32gb 的 RAM 和 4 个 CPU。还启用了火花推测。

这些任务的问题在哪里？ UDF有问题吗？是线程问题？

【问题讨论】：

你的分区数是多少？您是否尝试重新分区或更改 DataFrame 的分区数？也许您的分区不平衡：您在调用 udf 之前是否执行了可能导致分区不平衡的操作？
不平衡是什么意思？如何平衡数据框？在运行数据框之前，我已经使用 80000 个分区进行了重新分区。
不平衡是指某些分区，我的意思是您的 2900 万行没有在分区之间均匀拆分。你可以找到一些元素here。我认为这是很多分区的方式。如果我是你，我会尝试使用更少的分区。

标签： python apache-spark pyspark bigdata user-defined-functions

【解决方案1】：

我的直觉是你使用了太多的分区。我会通过显着减少他们的数量来进行第一次尝试。你可以在这个主题上找到这个有趣的post。

如果您的分区是平衡的，那么您按分区平均有 29 millions /80k partitions = 362 观察值。我想这还不够。您花费大量时间安排任务而不是执行任务。

如果您没有平衡分区，情况会变得更糟（请参阅here。这通常会造成瓶颈，这似乎是您的情况。有几个选项：

您可以coalesce您的数据到较少数量的分区。这比使用 repartition 更好，因为它避免了完全随机播放
repartitionByRange 如果您希望根据某些列拆分数据。您将不会像使用 coalesce 或 repartition 那样拥有平衡的分区，但后者可能很有用，因为您需要通过那些拆分列来使用操作

您可以使用spark.sql.shuffle.partitions 和spark.default.parallelism 更改有关分区的默认值。

这是根据我的经验猜测的。找到足够数量的分区很难，但值得。让我知道它是否有帮助或者您是否仍然遇到瓶颈。

【讨论】：

【解决方案2】：

我在第一个数据框中找到了使用 repartitionByRange 的解决方案。使用正确的 id 和分区数进行签名，它会平衡每个分区中的行数。

【讨论】：

所以很可能是分区不平衡的问题。我的回答对你有帮助吗？