python - 使用 Databricks 将文件从 Azure Blob 存储上传到 SFTP 位置？

Question

我有一个场景，需要将文件从 Azure Blob Storage 复制到 Databricks 中的 SFTP 位置

有没有办法使用pySpark或Scala来实现这个场景？

Answer 1

关于该问题，请引用以下步骤(我使用scala)

1. 将 Azure Blob 存储容器挂载到 DBFS

dbutils.fs.mount(
  source = "<container-name>@<storage-account-name>.blob.core.windows.net",
  mountPoint = "/mnt/blob",
  extraConfigs = Map("fs.azure.account.key.<storage-account-name>.blob.core.windows.net" -> "<key>"))

dbutils.fs.ls("/mnt/blob")

将这些文件复制到集群本地文件系统

%sh

cp -R /dbfs/mnt/blob  /databricks/driver
ls -R /databricks/driver/blob

代码。在运行代码之前，请在databricks中添加库com.jcraft.jsch vai Maven

import java.io.File
import scala.sys.process._
import com.jcraft.jsch._
def recursiveListFiles(f: File): Array[File] = {
  val these = f.listFiles
  these ++ these.filter(_.isDirectory).flatMap(recursiveListFiles)
}
val jsch = new JSch()
val session = jsch.getSession("<usename>", "<host>",<port>) // Set your username and host
session.setPassword("<password>") // Set your password
val config = new java.util.Properties()
config.put("StrictHostKeyChecking", "no")
session.setConfig(config)
session.connect()
val channelSftp = session.openChannel("sftp").asInstanceOf[ChannelSftp]
channelSftp.connect()

val files =recursiveListFiles(new File("/databricks/driver/blob"))

files.foreach(file =>{

  if(file.isFile()){
    println(file.getPath())
    channelSftp.put(file.getPath(),"/home/testqw/upload")
  }
     
  

})
channelSftp.disconnect()
session.disconnect()

请联系 FileZilla

<小时/>

#更新

挂载Azure blob后，我们可以直接访问文件并上传。

例如

import java.io.File
import scala.sys.process._
import com.jcraft.jsch._
def recursiveListFiles(f: File): Array[File] = {
  val these = f.listFiles
  these ++ these.filter(_.isDirectory).flatMap(recursiveListFiles)
}
val jsch = new JSch()
val session = jsch.getSession("", "",22) // Set your username and host
session.setPassword("") // Set your password
val config = new java.util.Properties()
config.put("StrictHostKeyChecking", "no")
session.setConfig(config)
session.connect()
val channelSftp = session.openChannel("sftp").asInstanceOf[ChannelSftp]
channelSftp.connect()
val home = channelSftp.getHome()

val files =recursiveListFiles(new File("/dbfs/mnt/blob"))

files.foreach(file =>{

  if(file.isFile()){
    println(file.getPath())
    channelSftp.put(file.getPath(),"/home/testqw/upload")
  }
     
  

})
channelSftp.disconnect()
session.disconnect()