本文旨在解决 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# 创建 SparkSessionspark = 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")# 停止 SparkSessionspark.stop()
注意事项
资源分配： 确保集群有足够的资源来满足你配置的 executor 数量和资源需求。网络带宽： 如果网络带宽是瓶颈，即使增加了并发请求的数量，写入速度也可能不会显著提高。S3 存储桶性能： S3 存储桶的性能也会影响写入速度。如果存储桶的写入能力有限，那么优化 Spark 和 AWS CLI 配置的效果也会受到限制。数据倾斜： 确保分区键能够均匀地分配数据，避免数据倾斜导致某些 executor 负载过重。
总结
通过显式设置 Spark executor 的数量和资源、优化 AWS CLI 配置，以及理解 Spark 任务分配机制，可以有效地解决 Spark 写入分区时仅使用单核的问题，从而提高写入速度，缩短整体作业的执行时间。 在实际应用中，需要根据具体情况调整配置参数，并监控 Spark UI 和 S3 存储桶的性能，以达到最佳的写入性能。
以上就是Spark 并行读取但写入分区时仅使用单核的优化方法的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1366824.html
微信扫一扫 
支付宝扫一扫 
