azure - 解压缩存储在 Azure Databricks FileStore 中的文件夹

Question

我已将 *.zip 文件夹上传到我的 Azure Datacricks FileStore:

现在我想解压该文件夹并将其存储在 FileStore 上:dbfs:/FileStore/tables/rfc_model。

我知道这应该很容易，但我在 DB Notebooks 中工作时感到困惑......

感谢您的帮助!

更新:

我使用这个公式没有成功:

%sh
解压/FileStore/tables/rfc_model.zip

和

%sh
解压dbfs:/FileStore/tables/rfc_model.zip

更新:

我已将 @Sim 创建的代码复制到我的 Databricks 笔记本中，但出现此错误:

知道如何解决这个问题吗？

Answer 1

当您使用 %sh 时，您正在使用其本地文件系统在驱动程序节点上执行 shell 命令。但是，/FileStore/ 不在本地文件系统中，这就是您遇到问题的原因。您可以通过尝试看到这一点:

%sh ls /FileStore
# ls: cannot access '/FileStore': No such file or directory

对比

dbutils.fs.ls("/FileStore")
// resX: Seq[com.databricks.backend.daemon.dbutils.FileInfo] = WrappedArray(...)

您必须使用可与 Databricks 文件系统配合使用的解压缩实用程序，或者必须将 zip 从文件存储复制到驱动程序磁盘，解压缩，然后复制回 /FileStore .

您可以使用file:/...寻址本地文件系统，例如，

dbutils.fs.cp("/FileStore/file.zip", "file:/tmp/file.zip")

希望这有帮助。

旁注 1:Databricks 文件系统管理并不是非常直观，尤其是在文件存储方面。例如，理论上，Databricks 文件系统 (DBFS) 在本地安装为 /dbfs/。但是，/dbfs/FileStore 不寻址文件存储，而dbfs:/FileStore 则寻址。你并不孤单。 :)

旁注 2:如果您需要对许多文件执行此操作，您可以通过使用文件路径和 ds 创建一个 Dataset[String] 将工作分配给集群工作人员.map { name => ... }.collect()。收集操作将强制执行。在 map 函数的主体中，您必须使用 shell API，而不是 %sh。

旁注 3:不久前，我使用以下 Scala 实用程序在 Databricks 上解压缩。无法验证它是否仍然有效，但它可以给您一些想法。

  def unzipFile(zipPath: String, outPath: String): Unit = {
    val fis = new FileInputStream(zipPath)
    val zis = new ZipInputStream(fis)
    val filePattern = """(.*/)?(.*)""".r
    println("Unzipping...")
    Stream.continually(zis.getNextEntry).takeWhile(_ != null).foreach { file =>
      // @todo need a consistent path handling abstraction
      //       to address DBFS mounting idiosyncracies
      val dirToCreate = outPath.replaceAll("/dbfs", "") + filePattern.findAllMatchIn(file.getName).next().group(1)
      dbutils.fs.mkdirs(dirToCreate)
      val filename = outPath + file.getName
      if (!filename.endsWith("/")) {
        println(s"FILE: ${file.getName} to $filename")
        val fout = new FileOutputStream(filename)
        val buffer = new Array[Byte](1024)
        Stream.continually(zis.read(buffer)).takeWhile(_ != -1).foreach(fout.write(buffer, 0, _))
      }
    }
  }