如何配置Spark Runner运行的Beam应用使用S3ACommitter并验证？

一、具体配置方法 #

要让Beam Spark Runner启用S3A Committer，需通过Spark配置传递关键参数，支持代码或命令行两种配置方式：

1. 代码中通过SparkPipelineOptions配置 #

SparkPipelineOptions options = PipelineOptionsFactory.as(SparkPipelineOptions.class);
// 启用Directory类型S3A Committer（通用原子提交实现）
options.getSparkConf().set("fs.s3a.committer.name", "directory");
// 临时目录冲突时直接替换
options.getSparkConf().set("fs.s3a.committer.staging.conflict-mode", "replace");
// 禁用魔法提交器，避免与Beam输出逻辑冲突
options.getSparkConf().set("fs.s3a.committer.magic.enabled", "false");
// 指定Spark SQL提交协议类
options.getSparkConf().set("spark.sql.sources.commitProtocolClass", 
                           "org.apache.spark.sql.execution.datasources.S3SQLCommitProtocol");
// 补充S3访问配置（按需调整密钥、端点等）
options.getSparkConf().set("fs.s3a.access.key", "your-access-key");
options.getSparkConf().set("fs.s3a.secret.key", "your-secret-key");

2. 命令行通过spark-submit传递配置 #

若通过spark-submit提交Beam应用，可在命令中添加--conf参数：

spark-submit \
  --class com.your.package.YourPipelineMain \
  --conf spark.hadoop.fs.s3a.committer.name=directory \
  --conf spark.hadoop.fs.s3a.committer.staging.conflict-mode=replace \
  --conf spark.hadoop.fs.s3a.committer.magic.enabled=false \
  --conf spark.sql.sources.commitProtocolClass=org.apache.spark.sql.execution.datasources.S3SQLCommitProtocol \
  --conf spark.hadoop.fs.s3a.access.key=your-access-key \
  --conf spark.hadoop.fs.s3a.secret.key=your-secret-key \
  your-beam-pipeline.jar

注意：确保hadoop-aws和hadoop-common版本与Spark依赖的Hadoop版本匹配（如Spark 3.3.x对应Hadoop 3.3.x），避免兼容性问题。

二、验证S3A Committer是否启用 #

1. 查看Spark作业日志 #

在Driver或Executor日志中搜索DirectoryCommitter或S3A Committer，出现以下日志则说明启用成功：

Using committer org.apache.hadoop.fs.s3a.commit.DirectoryCommitter

2. 检查S3输出目录行为 #

启用后数据会先写入临时staging目录（格式如.s3a-staging-xxxxxx），作业完成后原子性移动到目标目录。正常完成后临时目录会自动清理；若作业失败，可能残留临时目录，可作为启用的间接证据。

3. 查看Spark UI环境配置 #

打开Spark作业UI的Environment标签，在Spark Properties区域搜索fs.s3a.committer.name，确认值为directory（或你配置的类型），同时检查其他相关配置项是否生效。

内容的提问来源于stack exchange，提问作者Jie Jason Li