Spark 并行读取但写入分区时仅使用单核的优化方法-Python教程-PHP中文网

Spark 并行读取但写入分区时仅使用单核的优化方法

心靈之曲

发布： 2025-08-08 16:16:13

原创

789人浏览过

spark 并行读取但写入分区时仅使用单核的优化方法

本文旨在解决 Spark 在本地模式下读取 CSV 文件并写入 Iceberg 表时，读取阶段能够充分利用多核并行处理，而写入阶段却只能单核运行的问题。通过调整 Spark 配置、优化 AWS CLI 设置，以及理解 Spark 任务分配机制，帮助读者充分利用计算资源，提升 Spark 写入性能。

在使用 Spark 处理大数据集时，一个常见的瓶颈是写入数据的速度。尤其是在将数据写入到云存储（如 S3）时，如果写入过程没有充分利用可用的计算资源，会导致整体作业的执行时间显著增加。本文将深入探讨如何优化 Spark 写入操作，使其能够并行运行，从而提高写入速度。

理解问题：单核写入的原因

在本地模式下，即使指定了 local[*] 作为 master，Spark 仍然可能只使用一个核心进行写入操作。这通常是由于以下几个原因：

默认资源分配： Spark 的默认资源分配策略可能将所有任务分配给单个 executor，尤其是在本地模式下。
动态资源分配的误用： 开启 spark.dynamicAllocation.enabled 并不一定能解决问题，因为其资源分配依赖于默认的资源计算器，可能无法充分利用所有可用核心。
S3 写入限制： 默认的 AWS CLI 配置可能限制了并发请求的数量，从而导致写入速度受限。

解决方案：优化 Spark 配置

要解决单核写入的问题，需要调整 Spark 的配置，以确保任务能够并行执行。以下是一些建议的配置：

显式设置 Executor 数量和资源： 不要依赖动态资源分配，而是显式设置 executor 的数量、内存和核心数。
```
--master yarn \
--deploy-mode cluster \
--num-executors 4 \
--executor-memory 1G \
--executor-cores 1 \
--driver-memory 2G \
--driver-cores 1
```
登录后复制
- --num-executors: 指定要启动的 executor 数量。
- --executor-memory: 指定每个 executor 的内存大小。
- --executor-cores: 指定每个 executor 的核心数。
根据你的集群资源和数据规模，调整这些参数。
检查 Spark UI： 在写入操作开始时，通过 Spark History Server UI 检查 executor 和任务的数量，确保任务已正确分配到多个 executor。

优化 AWS CLI 配置

除了 Spark 配置，AWS CLI 的配置也会影响写入性能。可以通过增加并发请求的数量来提高写入速度。

修改 AWS CLI 配置文件： 修改 ~/.aws/config 文件，增加以下配置：
```
[default]
s3 =
    max_concurrent_requests = 20
    max_queue_size = 1000
    multipart_threshold = 64M
    multipart_chunksize = 16M
    max_bandwidth = 100MB/s
```
登录后复制
- max_concurrent_requests: 允许的最大并发请求数。
- max_queue_size: 任务队列的最大长度。
- multipart_threshold: 启用分段上传的最小文件大小。
- multipart_chunksize: 分段上传时每个分段的大小。
- max_bandwidth: 限制上传和下载的最大带宽。
根据你的网络环境和 S3 存储桶的性能，调整这些参数。

代码示例

以下是一个 Spark 写入 Iceberg 表的示例代码，包含了上述优化建议：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("IcebergWrite") \
    .master("yarn") \
    .config("spark.executor.memory", "1G") \
    .config("spark.executor.cores", "1") \
    .config("spark.num.executors", "4") \
    .config("spark.sql.catalog.my_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.hadoop.fs.s3a.fast.upload", "true") \
    .getOrCreate()

# 读取 CSV 文件
df = spark.read.csv("s3://my_bucket/my_data.csv", header=True, inferSchema=True)

# 数据转换和分区
df = df.withColumn("partition_key", df["some_column"] % 10)  # 示例分区键

# 写入 Iceberg 表
df.repartition("partition_key") \
    .write \
    .format("iceberg") \
    .mode("append") \
    .partitionBy(["partition_key"]) \
    .saveAsTable("glue_catalog.my_db.data")

# 停止 SparkSession
spark.stop()

登录后复制

注意事项

资源分配： 确保集群有足够的资源来满足你配置的 executor 数量和资源需求。
网络带宽： 如果网络带宽是瓶颈，即使增加了并发请求的数量，写入速度也可能不会显著提高。
S3 存储桶性能： S3 存储桶的性能也会影响写入速度。如果存储桶的写入能力有限，那么优化 Spark 和 AWS CLI 配置的效果也会受到限制。
数据倾斜： 确保分区键能够均匀地分配数据，避免数据倾斜导致某些 executor 负载过重。

总结

通过显式设置 Spark executor 的数量和资源、优化 AWS CLI 配置，以及理解 Spark 任务分配机制，可以有效地解决 Spark 写入分区时仅使用单核的问题，从而提高写入速度，缩短整体作业的执行时间。在实际应用中，需要根据具体情况调整配置参数，并监控 Spark UI 和 S3 存储桶的性能，以达到最佳的写入性能。

以上就是Spark 并行读取但写入分区时仅使用单核的优化方法的详细内容，更多请关注php中文网其它相关文章！