使用 Java 将文件写入 Google Cloud Storage 后临时文件不会被删除

Question

我正在使用 Hadoop GCS Connector 3-2.2.2 和 google-cloud-storage version 1.113.14 将数据写入 Google Cloud Storage 上的文件。

我有一个文件类，它有一个 write 方法，它使用 hadoop 类返回一个 OutputStream，如下所示：

protected FileSystem fs;
protected Path path;

public File(String path) {
    this.path = new Path(path);
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", fsURL);
    conf.setBoolean("dfs.support.append", true);
    conf.setInt("ipc.client.connect.max.retries.on.timeouts", 2);
    this.fs = FileSystem.newInstance(conf);
}

@Override
public OutputStream write(boolean overwriteIfExists) throws IOException {
    return fs.create(path, overwriteIfExists);
}

fs 是 org.apache.hadoop.fs.FileSystem 的实例； path 是 import org.apache.hadoop.fs.Path 的实例；

我什至尝试使用google cloud storage sdk来实现我的write方法：

@Override
public OutputStream write(boolean overwriteIfExists, boolean create) throws IOException {
    if (!overwriteIfExists && exists()) {
        throw new IOException("File already exists");
    }

    if (overwriteIfExists && exists()) {
        delete();
    } // If will write it will mean a new file

    BlobId blobId = BlobId.of(bucketName, objectName);
    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
    
    WriteChannel writeChannel = storage.writer(blobInfo);
    writeChannel.setChunkSize(64 * 1024 * 1024);

    return Channels.newOutputStream(writeChannel);
}

我的应用程序的其他部分可以创建我的 File 类的实例并写入 OutputStream，下面是如何测试/模拟写入的示例：

public void write() {
try {
  File file = File.newInstance("dir/someFileName");
  try (DataOutputStream os = new DataOutputStream(new BufferedOutputStream(file.write(true), this.bufferSize))) {
    long i;
    for (i = 0L; i < 4096; i++) {
      os.write(String.format("some arbitrary string %d\n", new Object[] { Long.valueOf(i) }).getBytes());
    } 
  } catch (Throwable t) {
    System.out.println("Couldn't write file");
  } 
} catch (IOException e) {
  System.out.println("Couldn't write file");
} 

}

当我使用gsutil ls dir/ 检查写入的文件后，我希望只找到我刚刚创建的 1 个文件，但我得到了 2 个应该被删除的额外临时文件（具有 GCS_SYNCABLE_TEMPFILE 前缀）

gs://dir/_GCS_SYNCABLE_TEMPFILE_someFileName.txt.0.24fccb78-8cea-45fd-852d-ed9385e4246b gs://dir/_GCS_SYNCABLE_TEMPFILE_someFileName.txt.0.cf132710-9d71-4939-a5d8-02b0279cdf3c gs://dir/someFileName.txt

我需要做一些配置吗？或者为什么这些临时文件没有被删除？

或者这可能是任何相关 jar 中的不兼容？我正在使用以下版本的 google 相关依赖项：

• 组：'org.apache.hadoop'，名称：'hadoop-common'，版本：'3.2.0'
• 组：'org.apache.hadoop'，名称：'hadoop-hdfs'，版本：'3.2.0'
• 组：'org.apache.hadoop'，名称：'hadoop-hdfs-client'，版本：'3.2.0'
• 组：“com.google.cloud.bigdataoss”，名称：“gcs-connector”，版本： ‘hadoop3-2.2.2’
• 组：“com.google.api-client”，名称： ‘google-api-client’，版本：‘1.31.3’
• 组：“com.google.cloud”， 名称：“google-cloud-core”，版本：“1.93.4”
• 组： ‘com.google.cloud’，名称：‘google-cloud-core-http’，版本：‘1.93.4’
• 组：“com.google.cloud”，名称：“google-cloud-storage”，版本： ‘1.106.0’
• 组：“com.google.http-client”，名称： ‘google-http-client’，版本：‘1.38.0’
• 组： 'com.google.http-client'，名称：'google-http-client-jackson2'， 版本：‘1.38.0’

Answer 1

• OutputStream 在提供的源代码中公开可用 这个GitHub link
• TEMP 文件的行为是客户端库的属性，而不是 直接 GCS。
• 随附的文档链接提供了详细信息，但一般 概述是：

在第一次调用 hsync()、sync() 或 close() 之前，此通道的行为方式与基本的不可同步通道相同，直接写入目标文件。 在第一次调用 hsync()/sync() 时，目标文件被提交，并使用隐藏文件前缀（下划线）创建一个新的临时文件，该文件带有一个附加后缀，该后缀对于系列中的每个后续临时文件都不同；在此期间，读者可以读取提交到目标文件的数据，但不能读取自​​上次调用 hsync() 后写入临时文件的字节。 在每个后续的 hsync()/sync() 调用中，临时文件 closed()，组合成目标文件，然后被删除，并以新文件名打开一个新的临时文件以供进一步写入。 警告：每个 hsync()/sync() 都需要顺序发生许多底层读取和突变请求，因此预计延迟会相当高。

如果中途出现错误，可能有一个或多个临时文件未能清理干净，需要人工干预才能发现并删除任何此类未使用的文件。在这种情况下，在最近一次成功的 hsync() 之前写入的数据是持久且安全的。

文件未关闭的可能原因是您没有关闭代码库中的流或出现错误。逐步调试您的代码以检查流是否正确关闭并删除文件，如图here

注意：由于代码是异步的，您需要确保执行删除的线程正在运行。