如何利用Spark从一个Minio集群读取数据并写入另一个Minio集群？

一、前置配置 #

Minio兼容S3协议，Spark通过Hadoop的S3AFileSystem对接Minio。由于要操作两个不同集群，需分别配置访问信息，避免全局覆盖：

在Spark配置中添加以下参数（Scala示例）：

val spark = SparkSession.builder()
  .appName("MinioCrossClusterTransfer")
  // 通用S3配置
  .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
  .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false") // Minio用HTTP时开启
  // 源Minio集群配置
  .config("spark.hadoop.fs.s3a.endpoint", "http://source-minio-cluster:9000")
  .config("spark.hadoop.fs.s3a.access.key", "SOURCE_ACCESS_KEY")
  .config("spark.hadoop.fs.s3a.secret.key", "SOURCE_SECRET_KEY")
  // 目标Minio集群（针对bucket单独配置，避免覆盖源集群）
  .config("spark.hadoop.fs.s3a.bucket.target-bucket.endpoint", "http://target-minio-cluster:9000")
  .config("spark.hadoop.fs.s3a.bucket.target-bucket.access.key", "TARGET_ACCESS_KEY")
  .config("spark.hadoop.fs.s3a.bucket.target-bucket.secret.key", "TARGET_SECRET_KEY")
  .getOrCreate()

二、读取源Minio数据 #

直接通过S3路径读取数据，支持Parquet、CSV、JSON等常见格式：

// 读取Parquet文件
val sourceDF = spark.read.parquet("s3a://source-bucket/input-data-path")

// 读取CSV文件（带表头）
// val sourceDF = spark.read.option("header", "true").csv("s3a://source-bucket/csv-input-path")

三、数据转换 #

根据业务需求编写转换逻辑，示例如下：

import org.apache.spark.sql.functions._

val transformedDF = sourceDF
  .filter($"user_status" === "active") // 过滤活跃用户
  .select($"user_id", $"user_name", $"register_date") // 保留指定字段
  .withColumn("process_timestamp", current_timestamp()) // 添加处理时间戳

四、写入目标Minio集群 #

指定目标bucket路径，设置写入模式（overwrite/append/ignore等）：

// 写入Parquet格式
transformedDF.write
  .mode("overwrite") // 根据需求选择写入模式
  .parquet("s3a://target-bucket/output-data-path")

// 写入CSV格式
// transformedDF.write
//   .mode("append")
//   .option("header", "true")
//   .csv("s3a://target-bucket/csv-output-path")

五、关键注意事项 #

• 依赖配置：确保Spark环境包含hadoop-aws和aws-java-sdk-bundle依赖，版本需与Spark的Hadoop版本匹配（如Spark 3.3对应Hadoop 3.3，依赖版本可选用hadoop-aws:3.3.4、aws-java-sdk-bundle:1.11.1026）
• 权限验证：确认源Minio账号拥有对应bucket的读取权限，目标Minio账号拥有对应bucket的写入权限
• 性能优化：大文件场景下，可调整spark.sql.shuffle.partitions参数设置并行度，提升处理效率

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