使用Spark的write()方法写入S3桶TXT文件失败，报错求助

问题背景 #

尝试通过Spark将Dataset<String>以TXT格式写入S3桶时，触发非法参数异常，提示必须指定AWS访问密钥ID和密钥。

错误信息 #

Exception in thread "main" java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.intellij.rt.execution.CommandLineWrapper.main(CommandLineWrapper.java:64)
Caused by: java.lang.IllegalArgumentException: AWS Access Key ID and Secret Access Key must be specified by setting the fs.s3.awsAccessKeyId and fs.s3.awsSecretAccessKey properties (respectively).

相关代码 #

Kotlin写入逻辑 #

override fun write(input: Dataset<String>) =
        input.coalesce(NUMBER_PARTITIONS).write().text(S3_BUCKET_PATH)
            .also {
                LOGGER.logInfo(
                    LOG_MESSAGE_TEMPLATE,
                    READ_DATA_METHOD,
                    WRITE_MESSAGE
                )
            }

Scala Spark配置类 #

object SparkConfiguration {
    private const val SPARK_MASTER_NAME = "spark.master"
    private const val SPARK_APP_NAME_CONFIG = "spark.app.name"
    fun buildSparkSession(config: Config): SparkSession {
        return SparkSession.builder()
            .config(buildSparkConfig(config))
            .orCreate
    }
    fun buildSparkConfig(config: Config): SparkConf = SparkConf()
        .setMaster(config.getString(SPARK_MASTER_NAME))
        .setAppName(config.getString(SPARK_APP_NAME_CONFIG))
}

解决方案 #

根据错误提示，需为Spark配置AWS访问密钥，以下是三种可行方式：

1. 在SparkConf中显式配置密钥 #

修改buildSparkConfig方法，从配置文件读取AWS密钥并添加到SparkConf：

fun buildSparkConfig(config: Config): SparkConf = SparkConf()
    .setMaster(config.getString(SPARK_MASTER_NAME))
    .setAppName(config.getString(SPARK_APP_NAME_CONFIG))
    // 添加AWS密钥配置
    .set("fs.s3.awsAccessKeyId", config.getString("aws.access.key.id"))
    .set("fs.s3.awsSecretAccessKey", config.getString("aws.secret.access.key"))

同时确保配置文件中包含aws.access.key.id和aws.secret.access.key字段，值为你的AWS账号密钥。

2. 通过环境变量配置 #

在运行Spark程序的环境中设置以下环境变量，Spark会自动读取这些值：

export AWS_ACCESS_KEY_ID="你的Access Key ID"
export AWS_SECRET_ACCESS_KEY="你的Secret Access Key"

3. 使用IAM角色（AWS环境专属） #

如果Spark程序运行在AWS EMR、EC2等服务上，可为实例绑定具备S3写入权限的IAM角色，无需手动配置密钥，Spark会自动通过IAM角色获取访问权限。

注意：若使用S3A文件系统（较新版本Spark推荐），配置项需改为fs.s3a.access.key和fs.s3a.secret.key，请根据所用S3客户端调整配置键名。

内容的提问来源于stack exchange，提问作者patrick pascoal ribeiro