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

【问题标题】：Uploading files from Azure Blob Storage to SFTP location using Databricks?使用 Databricks 将文件从 Azure Blob 存储上传到 SFTP 位置？
【发布时间】：2020-09-08 11:46:07
【问题描述】：

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

有没有办法使用pySpark 或Scala 来实现这种情况？

【问题讨论】：

标签： python scala azure pyspark azure-databricks

【解决方案1】：

关于问题，请参考以下步骤（我使用scala）

将 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中添加librarycom.jcraft.jschvai 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()

【讨论】：

@DipanjanMallick 如果我们使用com.jcraft.jsch 上传文件，我们无法从挂载位置访问文件。如果你想从挂载的位置访问文件，我们只能用 spark 读取文件，然后我们可以使用 spark sftp 连接器上传文件。但它有一些限制。我们应该使用 spark 2.x 和 scala 2.1.1：github.com/springml/spark-sftp
感谢@Jim Xu 的更新......在这方面不确定！因为我可以直接从安装位置上传文件
@DipanjanMallickd 你能告诉我你是怎么做到的吗？
我刚刚将 /databricks/driver/ 替换为 /dbfs/mnt/blob 如下 val files =recursiveListFiles(new File(" /dbfs/mnt/blob"))
@DipanjanMallick 感谢您的分享。如果您的问题已经解决，您能否接受它作为答案？