从SFTP服务器加载文件到spark RDD
问题描述:
如何从SFTP服务器加载文件到spark RDD。加载这个文件后,我需要对数据执行一些过滤。此外,该文件是csv文件,所以你可以帮我决定是否应该使用数据框或RDD。从SFTP服务器加载文件到spark RDD
答
您可以以下方式使用spark-sftp
库在你的程序:
火花2.x的
Maven的依赖
<dependency>
<groupId>com.springml</groupId>
<artifactId>spark-sftp_2.11</artifactId>
<version>1.1.0</version>
</dependency>
SBT依赖
libraryDependencies += "com.springml" % "spark-sftp_2.11" % "1.1.0"
与火花壳
此包可添加使用--packages命令行选项来使用火花。
$ bin/spark-shell --packages com.springml:spark-sftp_2.11:1.1.0
Scala的API
// Construct Spark dataframe using file in FTP server
val df = spark.read.
format("com.springml.spark.sftp").
option("host", "SFTP_HOST").
option("username", "SFTP_USER").
option("password", "****").
option("fileType", "csv").
option("inferSchema", "true").
load("/ftp/files/sample.csv")
// Write dataframe as CSV file to FTP server
df.write.
format("com.springml.spark.sftp").
option("host", "SFTP_HOST").
option("username", "SFTP_USER").
option("password", "****").
option("fileType", "csv").
save("/ftp/files/sample.csv")
用于火花的1.x(1.5+)
Maven的依赖
:例如,要在启动火花壳时它包括<dependency>
<groupId>com.springml</groupId>
<artifactId>spark-sftp_2.10</artifactId>
<version>1.0.2</version>
</dependency>
SBT依赖
libraryDependencies += "com.springml" % "spark-sftp_2.10" % "1.0.2"
与火花壳
此包可添加使用--packages
命令行选项来使用火花。例如,在启动火花外壳时,它包含:
$ bin/spark-shell --packages com.springml:spark-sftp_2.10:1.0.2
斯卡拉API
import org.apache.spark.sql.SQLContext
// Construct Spark dataframe using file in FTP server
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.
format("com.springml.spark.sftp").
option("host", "SFTP_HOST").
option("username", "SFTP_USER").
option("password", "****").
option("fileType", "csv").
option("inferSchema", "true").
load("/ftp/files/sample.csv")
// Write dataframe as CSV file to FTP server
df.write().
format("com.springml.spark.sftp").
option("host", "SFTP_HOST").
option("username", "SFTP_USER").
option("password", "****").
option("fileType", "csv").
save("/ftp/files/sample.csv")
欲了解更多有关spark-sftp
你可以参观那里的github页springml/spark-sftp
答
加载从SFTP是直接使用sftp连接器。
https://github.com/springml/spark-sftp
记住它是单线程的应用程序和数据的土地到HDFS即使你不指定它。它将数据流化为hdfs,然后在其上创建一个DataFrame
加载时,我们需要指定几个参数。
通常在没有指定位置的情况下,当用户sudo用户使用hdfs时,它也可能正常工作。它将在/ hdfs中创建临时文件,并在完成该过程后将其删除。
val data = sparkSession.read.format("com.springml.spark.sftp").
option("host", "host").
option("username", "user").
option("password", "password").
option("fileType", "json").
option("createDF", "true").
option("hdfsTempLocation","/user/currentuser/").
load("/Home/test_mapping.json");
所有可用的选项下面,源代码
override def createRelation(sqlContext: SQLContext, parameters: Map[String, String], schema: StructType) = {
val username = parameters.get("username")
val password = parameters.get("password")
val pemFileLocation = parameters.get("pem")
val pemPassphrase = parameters.get("pemPassphrase")
val host = parameters.getOrElse("host", sys.error("SFTP Host has to be provided using 'host' option"))
val port = parameters.get("port")
val path = parameters.getOrElse("path", sys.error("'path' must be specified"))
val fileType = parameters.getOrElse("fileType", sys.error("File type has to be provided using 'fileType' option"))
val inferSchema = parameters.get("inferSchema")
val header = parameters.getOrElse("header", "true")
val delimiter = parameters.getOrElse("delimiter", ",")
val createDF = parameters.getOrElse("createDF", "true")
val copyLatest = parameters.getOrElse("copyLatest", "false")
//System.setProperty("java.io.tmpdir","hdfs://devnameservice1/../")
val tempFolder = parameters.getOrElse("tempLocation", System.getProperty("java.io.tmpdir"))
val hdfsTemp = parameters.getOrElse("hdfsTempLocation", tempFolder)
val cryptoKey = parameters.getOrElse("cryptoKey", null)
val cryptoAlgorithm = parameters.getOrElse("cryptoAlgorithm", "AES")
val supportedFileTypes = List("csv", "json", "avro", "parquet")
if (!supportedFileTypes.contains(fileType)) {
sys.error("fileType " + fileType + " not supported. Supported file types are " + supportedFileTypes)
}
什么火花spark.read?它是sc吗?它给我错误。任何导入它? – vindev
'spark.read'中的spark是'SparkSession',它是在Spark 2.0中引入的。如果您使用的是Spark版本 himanshuIIITian
谢谢:)其复制文件,但在加载语句中给出以下错误。 17/04/14 12:51:17 INFO SFTPClient:成功复制文件... 线程“main”中的异常java.lang.NoSuchMethodError:org.apache.spark.sql.DataFrameReader.load(Ljava/lang/String ;)Lorg /阿帕奇/火花/ SQL /数据集; – vindev