如何优化 Spark 驱动程序的内存使用？使用单一 Delta 表格和分区

python

2024-03-25 03:47:50

优化 Spark 驱动程序内存使用量：使用单一 Delta 表格和分区

问题：

在运行结构化 Spark 查询时，许多人遇到了 Spark 驱动程序占用大量内存的问题。此问题尤其发生在处理从 Kafka 流中提取的数据并将其写入 S3 中的 Delta 表格时。

根本原因：

问题根源于创建多个小型 Delta 表格以存储属于不同对话的 message 对。这导致 Spark 驱动程序消耗过量内存，影响查询性能。

解决方案：

为了解决此问题，我们建议采用以下优化措施：

使用单一 Delta 表格

避免创建多个 Delta 表格。相反，使用一个单一的 Delta 表格并追加数据。这可以显著减少内存使用量。

应用分区

使用对话 ID 对 Delta 表格进行分区。这有助于提高查询性能，因为 Spark 可以只扫描特定对话的分区，而不是整个表。

优化后的代码

以下是优化后的 write_to_s3 函数代码示例：

def write_to_s3(spark: SparkSession, s3_path: str, batch_df: DataFrame, _: int) -> None:
    # 创建单一 Delta 表格
    delta_path = os.path.join(s3_path, "turn_history")
    if not DeltaTable.isDeltaTable(spark, delta_path):
        batch_df.write.format("delta").option("path", delta_path).saveAsTable(
            TABLE_NAME
        )

    # 追加数据到 Delta 表格
    delta_table = DeltaTable.forPath(spark, delta_path)
    delta_table.alias("turns").merge(
        batch_df.alias("newTurns"),
        "turns.turn_id = newTurns.turn_id",
    ).whenNotMatchedInsertAll().execute()