使用Spark将Azure Blob容器src的镜像文件复制到dst容器

方案1：读取二进制文件后写入目标路径 #

这种方式直接借助Spark的二进制文件读写能力，将源文件内容读取后写入目标路径，适合文件数量适中、大小不算超大的场景，天然支持分布式处理。

操作步骤： #

假设存储路径映射的DataFrame结构为src_path: String, dst_path: String：

1. 读取源文件的二进制内容：

   import org.apache.spark.sql.functions._
   
   // 假设映射DataFrame名为path_mapping_df
   val binary_content_df = path_mapping_df
     .select("src_path")
     .distinct()
     .withColumn("content", binary_file(col("src_path")))
   

2. 关联映射关系并写入目标路径：

   binary_content_df
     .join(path_mapping_df, Seq("src_path"), "inner")
     .select("dst_path", "content")
     .write
     .mode("overwrite")
     .binaryFile()
   

   注：此方式会将文件内容加载到Executor内存中，超大文件（如GB级）可能引发内存溢出问题，需根据文件大小调整Executor内存配置。

方案2：通过UDF调用Hadoop FileSystem API实现分布式复制 #

利用Hadoop底层的FileSystem API编写Spark UDF，实现完全分布式的文件复制，适合大量文件或超大文件场景，性能远优于单节点的dbutils.fs.cp，且可在Spark函数中正常使用。

操作步骤： #

1. 编写文件复制UDF（Scala示例）：

   import org.apache.hadoop.fs.{FileSystem, Path}
   import org.apache.spark.sql.functions.udf
   
   // 基于Spark的Hadoop配置初始化文件系统客户端
   val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
   
   // 定义UDF：输入源路径和目标路径，返回复制是否成功
   val copyFileUdf = udf((srcPathStr: String, dstPathStr: String) => {
     val srcPath = new Path(srcPathStr)
     val dstPath = new Path(dstPathStr)
     if (fs.exists(srcPath)) {
       // 参数含义：是否删除源文件、是否覆盖目标文件、源路径、目标路径
       fs.copyFromLocalFile(false, true, srcPath, dstPath)
       true
     } else {
       false
     }
   })
   

   注：DBFS挂载的路径（如/dbfs/mnt/src/xxx.img）可被Hadoop FileSystem直接识别，无需额外配置。
2. 执行分布式复制：

   // 触发复制任务（执行count()等action操作才会实际运行）
   val copy_result_df = path_mapping_df
     .withColumn("copy_success", copyFileUdf(col("src_path"), col("dst_path")))
   copy_result_df.count()
   

关键注意事项 #

• 权限验证：确保Spark使用的服务主体已拥有src容器的读权限和dst容器的写权限，Hadoop FileSystem会自动继承Spark的身份认证信息。
• 覆盖控制：根据业务需求调整fs.copyFromLocalFile的overwrite参数（第二个布尔值），避免误删已有文件。
• 性能对比：dbutils.fs.cp为单节点串行执行，而上述两种Spark方案均为分布式并行处理，在文件数量较多时性能提升显著，方案2的流式复制对大文件更友好。

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