Spark如何实现同一S3存储桶不同路径使用多AWS凭证？

问题背景 #

• 需求：关联同一S3存储桶下不同路径的表，各路径表分属不同所有者，拥有独立的AWS临时凭证（访问密钥/密钥/会话令牌），数据及访问方式不可修改
• 现有局限：Spark支持按存储桶配置凭证，但无法直接应用于同一存储桶多路径场景
• 当前低效方案：将所有表下载至Spark驱动端，执行器从驱动端读取，存在驱动端瓶颈问题

尝试过的方法及问题 #

尝试通过修改Hadoop配置切换凭证，代码如下：

hadoopConf = spark._jsc.hadoopConfiguration() 
hadoopConf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") 
hadoopConf.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider") 
hadoopConf.set("fs.s3a.access.key", '***') 
hadoopConf.set("fs.s3a.secret.key", '***') 
hadoopConf.set("fs.s3a.session.token", "***")

成功读取表1后，用同样方法更新为表2的凭证，但配置未生效，访问表2时仍使用表1的凭证。

可行优化方案 #

1. S3A路径别名+独立凭证配置（推荐，Hadoop 3.3+） #

利用Hadoop的S3A路径别名功能，给同一存储桶的不同路径配置独立别名，每个别名绑定一套凭证：

• 配置示例（可通过SparkConf或hadoop-site.xml设置）：

  # 基础存储桶配置
  fs.s3a.bucket.my-bucket.path.style.access=true
  
  # 表1凭证配置
  fs.s3a.alias.table1.aws.credentials.provider=org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider
  fs.s3a.alias.table1.access.key=TABLE1_ACCESS_KEY
  fs.s3a.alias.table1.secret.key=TABLE1_SECRET_KEY
  fs.s3a.alias.table1.session.token=TABLE1_SESSION_TOKEN
  
  # 表2凭证配置
  fs.s3a.alias.table2.aws.credentials.provider=org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider
  fs.s3a.alias.table2.access.key=TABLE2_ACCESS_KEY
  fs.s3a.alias.table2.secret.key=TABLE2_SECRET_KEY
  fs.s3a.alias.table2.session.token=TABLE2_SESSION_TOKEN
  

• 读取数据时使用别名路径：

  df1 = spark.read.parquet("s3a://table1/my-bucket/path/to/table1")
  df2 = spark.read.parquet("s3a://table2/my-bucket/path/to/table2")
  joined_df = df1.join(df2, on="join_key")
  

  该方案无需修改代码逻辑，执行器直接从S3读取数据，避免驱动端瓶颈。

2. 多Spark会话分别配置凭证 #

为每个路径创建独立的Spark会话，各自绑定对应凭证：

# 会话1：读取表1
spark1 = SparkSession.builder \
    .config("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider") \
    .config("fs.s3a.access.key", "TABLE1_ACCESS_KEY") \
    .config("fs.s3a.secret.key", "TABLE1_SECRET_KEY") \
    .config("fs.s3a.session.token", "TABLE1_SESSION_TOKEN") \
    .getOrCreate()
df1 = spark1.read.parquet("s3a://my-bucket/path/to/table1")

# 会话2：读取表2
spark2 = SparkSession.builder \
    .config("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider") \
    .config("fs.s3a.access.key", "TABLE2_ACCESS_KEY") \
    .config("fs.s3a.secret.key", "TABLE2_SECRET_KEY") \
    .config("fs.s3a.session.token", "TABLE2_SESSION_TOKEN") \
    .getOrCreate()
df2 = spark2.read.parquet("s3a://my-bucket/path/to/table2")

# 关联数据
joined_df = df1.join(df2, on="join_key")

注意：多会话会增加资源开销，需根据集群资源情况评估使用。

3. 自定义路径感知凭证提供器 #

实现自定义AWSCredentialsProvider类，根据请求的S3路径动态返回对应凭证：

• 核心逻辑：在getCredentials()方法中解析当前访问的S3路径前缀，匹配并返回对应凭证
• 配置方式：

  hadoopConf.set("fs.s3a.aws.credentials.provider", "com.yourcompany.CustomPathCredentialsProvider")
  

  该方案灵活性最高，但需要开发和维护自定义类，且需确保类在Spark驱动和执行器的类路径中。

内容的提问来源于stack exchange，提问作者bigfox