scala - 从 SFTP 服务器加载文件到 spark RDD

Question

如何将文件从 SFTP 服务器加载到 spark RDD。加载此文件后，我需要对数据执行一些过滤。该文件也是 csv 文件，所以请您帮我决定是否应该使用 Dataframes 或 RDDs。

Answer 1

您可以使用 spark-sftp通过以下方式在您的程序中添加库:

对于 Spark 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"

与 Spark shell 一起使用

可以使用 --packages 命令行选项将此包添加到 Spark。例如，在启动 spark shell 时包含它:

$ 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")

对于 Spark 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"

与 Spark shell 一起使用

可以使用 --packages 将此包添加到 Spark。命令行选项。例如，在启动 spark shell 时包含它:

$ bin/spark-shell --packages com.springml:spark-sftp_2.10:1.0.2

Scala 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