CDAP源插件从Sftp服务器读取数据

Question

我想通过使用cdap源代码插件来读取Sftp服务器可用的csv文件。

我偶然发现FTP批处理源插件也是如此。但是，在运行这个程序时，我会遇到异常。

Caused by: java.io.IOException: No FileSystem for scheme: sftp
    at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2798) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2809) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:100) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2848) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2830) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:389) ~[org.apache.hadoop.hadoop-common-2.8.0.jar:na]
    at co.cask.hydrator.format.plugin.AbstractFileSource.prepareRun(AbstractFileSource.java:129) ~[na:na]
    at co.cask.hydrator.format.plugin.AbstractFileSource.prepareRun(AbstractFileSource.java:63) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource$1.call(WrappedBatchSource.java:53) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource$1.call(WrappedBatchSource.java:50) ~[na:na]
    at co.cask.cdap.etl.common.plugin.Caller$1.call(Caller.java:30) ~[na:na]
    at co.cask.cdap.etl.common.plugin.StageLoggingCaller.call(StageLoggingCaller.java:40) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource.prepareRun(WrappedBatchSource.java:50) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource.prepareRun(WrappedBatchSource.java:36) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource$1.call(WrappedBatchSource.java:53) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource$1.call(WrappedBatchSource.java:50) ~[na:na]
    at co.cask.cdap.etl.common.plugin.Caller$1.call(Caller.java:30) ~[na:na]
    at co.cask.cdap.etl.common.plugin.StageLoggingCaller.call(StageLoggingCaller.java:40) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource.prepareRun(WrappedBatchSource.java:50) ~[na:na]
    at co.cask.cdap.etl.common.plugin.WrappedBatchSource.prepareRun(WrappedBatchSource.java:36) ~[na:na]
    at co.cask.cdap.etl.common.submit.SubmitterPlugin$3.run(SubmitterPlugin.java:83) ~[na:na]
    at co.cask.cdap.internal.app.runtime.AbstractContext$2.run(AbstractContext.java:534) ~[na:na]
    at co.cask.cdap.data2.transaction.Transactions$CacheBasedTransactional.finishExecute(Transactions.java:224) ~[na:na]
    ... 18 common frames omitted

我正在使用以下版本的图书馆，这也是一个限制。

1. Hadoop - 2.7.3
2. Spark - 2.3.0

我还遇到了这问题，它建议使用这并将proeprty fs.sftp.impl设置为org.apache.hadoop.fs.sftp.SFTPFileSystem将解决这个问题，但不确定如何使用上述代码并设置这个proeprty。

Answer 1

使用SFTP作为协议时，需要在Advanced部分下设置文件系统属性：

{
  "fs.sftp.impl": "org.apache.hadoop.fs.sftp.SFTPFileSystem"
}

Answer 2

对于我的理解，FTP插件是不可取的。改用SFTP动作插件，并构建如下管道：

​

​

其想法是首先将文件复制到运行时环境的本地文件系统，然后在您想要的地方接收该文件(在我的例子中是GCS)。