如何删除Spark Scala中的实木复合地板文件？

如何解决如何删除Spark Scala中的实木复合地板文件？

我正在用数据块写入一个镶木地板文件，但在此之前，我想删除它的旧版本。

这是我的写信行：

report.coalesce(1).write.mode("append").partitionBy("Name").parquet(s"s3://${reportBucket}/reports/dashboard")

我不知道如何检查此文件的存在，并删除该文件是否存在。

该类的一些伪代码以及调用它的代码行。我正在尝试检查输出文件是否存在（如果存在），然后将其删除。删除该类后，该类将运行两次，并将结果附加到镶木地板文件中。但是，只能在run2之后删除它，而不能在run1之后删除。

class WriteReport(val run: String = "run1") {
val report = spark.read.parquet(s"blablah")
report.createOrReplaceTempView("report")

val dashboard = spark.sql (""" 
                SELECT name as Name from Table
                """)

report.coalesce(1).write.mode("append").partitionBy("Name").parquet(s"s3://${reportBucket}/reports/dashboard")
}

val n_b = new Report ("run1")
val n_g = new Report ("run2")

解决方法

Spark不提供对S3 delete的支持，您只能在 框架和其他外部框架任务，您需要建立自己的逻辑 在Spark之外。在触发EMR之前，您需要将lamda配置为 清理目录并确认删除后触发Spark Job。

在重新创建文件之前删除本地目录：

    import scala.reflect.io.Directory
    import java.io.File

    val dir = new Directory(new File("/yourDirectory"))
    dir.deleteRecursively()

在重新创建文件之前删除AWS S3中的目录：

AWS S3在PUTS对象上具有读写后一致性，最终一致性会覆盖PUTS和DELETES对象，因此一旦删除将不会删除相同的时间，并且会花费一些时间

因此，使用S3公共域并删除并同时运行新作业会遇到一些问题，您需要构建逻辑以每次创建单独的目录并在工作完成后删除。

Java：

您需要使用AWS sdk进行删除，因为Spark不支持从S3中删除任何命令

    if (s3Client.doesBucketExist(bucketName)) {
                ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
                        .withBucketName(bucketName)
                        .withPrefix("foo/bar/baz");

                ObjectListing objectListing = s3Client.listObjects(listObjectsRequest);

                while (true) {
                    for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries()) {
                        s3Client.deleteObject(bucketName,objectSummary.getKey());
                    }
                    if (objectListing.isTruncated()) {
                        objectListing = s3Client.listNextBatchOfObjects(objectListing);
                    } else {
                        break;
                    }
                }
            }

BOTO S3 SDK：

     import boto3

     client = boto3.client('s3')
    client.delete_object(Bucket='mybucketname',Key='myfile.whatever')