spark读取oss数据写入另一个oss 使用scala实现

以下是使用Scala实现Spark读取OSS数据并写入另一个OSS的示例代码：

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import com.aliyun.oss.OSSClient
import com.aliyun.oss.model.ObjectMetadata
import com.aliyun.oss.model.PutObjectResult

object OSSReadWriteExample {
  def main(args: Array[String]): Unit = {
    val accessKeyId = "your-access-key-id"
    val accessKeySecret = "your-access-key-secret"
    val endpoint = "your-oss-endpoint"
    val inputBucketName = "your-input-bucket-name"
    val outputBucketName = "your-output-bucket-name"
    val inputPath = "your-input-file-path"
    val outputPath = "your-output-file-path"

    // Create SparkSession
    val sparkConf = new SparkConf().setAppName("OSS Read Write Example")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // Read data from OSS
    val inputData = spark.read.textFile(s"oss://${inputBucketName}.${endpoint}/${inputPath}")

    // Process data
    val processedData = inputData.map(line => line.toUpperCase())

    // Write data to OSS
    val ossClient = new OSSClient(endpoint, accessKeyId, accessKeySecret)
    val outputObjectMetadata = new ObjectMetadata()
    outputObjectMetadata.setContentLength(processedData.toString().length)
    val outputObject = new PutObjectResult()
    val outputStream = processedData.map(line => line + "\n").map(_.getBytes("UTF-8")).collect().flatten
    ossClient.putObject(outputBucketName, outputPath, new java.io.ByteArrayInputStream(outputStream), outputObjectMetadata)

    // Close OSS client and SparkSession
    ossClient.shutdown()
    spark.close()
  }
}

在代码中，首先设置了OSS的访问凭证、OSS的Endpoint、输入和输出的Bucket名称、输入和输出的文件路径。然后创建了一个SparkSession，读取了输入文件中的数据，对数据进行处理，最后将处理后的结果写入到输出文件中。在写入数据到OSS时，使用了阿里云Java SDK提供的OSSClient类，创建了一个OSS客户端，通过putObject方法将数据写入到指定的OSS Bucket和文件路径中。最后关闭了OSS客户端和SparkSession。

需要注意的是，使用阿里云Java SDK时需要将对应的jar包添加到项目中，例如：

<dependency>
    <groupId>com.aliyun.oss</groupId>
    <artifactId>aliyun-sdk-oss</artifactId>
    <version>3.11.0</version>
</dependency>

同时，需要在pom.xml中添加OSS Maven仓库的配置：

<repositories>
    <repository>
        <id>oss-aliyun</id>
        <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
    </repository>
</repositories>