优化 Spark 读写性能

Question

我有大约 12K 个二进制文件，每个文件大小为 100mb，并且包含多个具有 变量 长度的压缩记录。我试图找到最有效的方法来读取它们，解压缩然后以镶木地板格式写回。我拥有的集群是 6 个节点，每个节点有 4 个核心。

此时使用下面的伪代码，读取所有文件大约需要 8 小时，写回 parquet 非常慢。

def reader(file_name):
    keyMsgList = []
    with open(file_name, "rb") as f:
        while True:
            header = f.read(12)
            if not header:
                break
            keyBytes = header[0:8]
            msgLenBytes = header[8:12]

            # conver keyBytes & msgLenBytes to int 
            message = f.read(msgLen)
            keyMsgList.append((key, decode(message)))
    return keyMsgList
files = os.listdir("/path/to/binary/files")
rddFiles = sc.parallelize(files, 6000)
df = spark.createDataFrame(rddFiles.flatMap(reader), schema)
df.repartition(6000).write.mode("append").partitionBy("key").parquet("/directory")

在这里选择 6000 sc.parallelize(files, 6000) 的原因是创建每个大小为 200 MB 的分区，即 (12k files * 100mb size) / 200MB。作为需要逐字节读取每个文件内容的顺序性质，不确定是否可以进一步优化读取？ 同样，写回parquet时，repartition(6000)中的数字是为了保证数据分布均匀，所有executor可以并行写。但是，结果证明这是一个非常缓慢的操作。

一种解决方案是增加执行器的数量，这会提高读取性能但不确定是否会提高写入？

在此处寻找有关如何提高性能的任何建议？

Answer 1

建议一：不要使用repartition，而是使用coalesce。

见here。您确定了repartition 操作的瓶颈，这是因为您已经启动了一个完整的洗牌。使用coalesce 你不会那样做。您最终也会得到 N 个分区。它们不会像 repartition 那样平衡，但这有关系吗？

我建议你支持coalesce 而不是repartition

建议 2：6000 个分区可能不是最优的

您的应用程序使用 6 个节点和 4 个内核运行。您有 6000 个分区。这意味着您有大约 250 个核心分区（甚至不包括提供给您的主服务器的分区）。在我看来，这太过分了。

由于您的分区很小（大约 200Mb），您的 master 可能会花费更多时间等待执行程序的 anwsers，而不是执行查询。

我会建议你减少分区的数量

建议 3：可以使用 DataFrame API 吗？

DataFrame API 操作通常比手动编码的解决方案更快更好。

也许看看pyspark.sql.functions 看看你是否可以在那里找到一些东西（见here）。我不知道这是否相关，因为我没有看到您的数据，但这是我根据经验提出的一般建议。