有啥方法可以提高 PySpark 输出的效率吗？

Posted 2023-04-15

【中文标题】有啥方法可以提高 PySpark 输出的效率吗？

【英文标题】：Is there any way to increase the efficiency of PySpark outputs?有什么方法可以提高 PySpark 输出的效率吗？

【发布时间】：2017-12-01 18:30:08

【问题描述】：

我正在尝试测试 PySpark 迭代一些非常大的数据（10 GB 到 1 TB）数据的能力。对于大多数脚本，我发现 PySpark 的效率与 Scala 代码大致相同。在其他情况下（如下面的代码），我会遇到严重的速度问题，速度会慢 10 到 12 倍。

path = "path/to/file"
spark = SparkSession.builder.appName("siteLinkStructureByDate").getOrCreate()
sc = spark.sparkContext   

df = RecordLoader.loadSomethingAsDF(path, sc, spark)
fdf = df.select(df['aDate'], df['aSourceUrl'], df['contentTextWithUrls'])
rdd = fdf.rdd
rddx = rdd.map (lambda r: (r.aDate, CreateAVertexFromSourceUrlAndContent(r.aSourceUrl, r.contentTextWithUrls)))\
 .flatMap(lambda r: map(lambda f: (r[0], ExtractDomain(f[0]), ExtractDomain(f[1])), r[1]))\
 .filter(lambda r: r[-1] != None)\
 .countByValue()

print([((x[0], x[1], x[2]), y) for x, y in rddx.items()]) 

我们认为我们已将问题隔离到 .countByValue()（它返回一个 defaultdict），但应用 countItems() 或 reduceByKey() 会产生几乎相同的结果。我们也 99% 确定问题不在于 ExtractDomain 或 CreateAVertexFromSourceUrlAndContent（不是函数的真实名称，只是为了使其易于理解的伪代码）。

所以我的问题是第一个

我可以在这段代码中做些什么来减少时间吗？ PySpark 从根本上说是 比它的 Scala 慢得多吗 对方？ 有没有办法复制平面图 改用 PySpark 数据帧（了解数据帧是 通常比 Pyspark 中的 RDD 快）？

【参考方案1】：

这里最大的问题可能是通信 - Spark SQL（列格式） -> 普通 Scala 对象 -> pickle (Pyrolite) -> 套接字 -> unpickle -> 普通 Python 对象。这需要大量的复制、转换和移动。

有一种方法可以使用 PySpark 数据帧来复制平面地图

是的。它被称为explode - 但公平地说，它也很慢。

了解数据帧通常比 Pyspark 中的 RDD 更快

这通常是正确的（Scala 和 Python 都是），但您可能需要 udf 来实现 ExtractDomain 或 CreateAVertexFromSourceUrlAndContent - 这是另一件缓慢的事情。仅从您可能可以使用parse_url_tuple 的名称来看。

从根本上说，PySpark 是否比它的 Scala 慢很多？

它有点慢。通常在经过良好调整的代码上不会那么慢。但实现细节有所不同——Scala 和 Python 中的同一组操作可以以不同的方式实现。

我可以在这段代码中做些什么来减少时间吗？

我建议先进行分析。一旦确定了负责的部分（转换、合并），您就可以尝试定位它。