python - 我如何在独立的 spark 集群 (pySpark) 中使用 FTP 上的文件？

Question

嘿，我是 spark 的新手，最近用几台笔记本电脑设置了一个 spark 独立集群。

我在我的本地 ftp 服务器上共享了一个名为 new.txt 的文件根据 Spark 指南“PySpark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3 等”( http://spark.apache.org/docs/latest/programming-guide.html#external-datasets )

我通过执行以下操作在独立集群模式下打开了 pyspark 交互式 shell:

 $ MASTER=spark://IP:PORT ./bin/pyspark

然后执行指南中的示例命令。

>>> ff= sc.textFile("ftp://192.168.125.124/new.txt")
>>> ans = ff.map(lambda s: len(s)).reduce(lambda a, b: a + b)

第一行执行正常并创建了 rdd。在第二行之后我收到了这个错误:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/shrey/spark-1.6.1/python/pyspark/rdd.py", line 797, in reduce
    vals = self.mapPartitions(func).collect()
  File "/home/shrey/spark-1.6.1/python/pyspark/rdd.py", line 771, in collect
    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
  File "/home/shrey/spark-1.6.1/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py", line 813, in __call__
  File "/home/shrey/spark-1.6.1/python/pyspark/sql/utils.py", line 45, in deco
    return f(*a, **kw)
  File "/home/shrey/spark-1.6.1/python/lib/py4j-0.9-src.zip/py4j/protocol.py", line 308, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: java.io.IOException: Login failed on server - 192.168.125.124, port - 21
    at org.apache.hadoop.fs.ftp.FTPFileSystem.connect(FTPFileSystem.java:133)
    at org.apache.hadoop.fs.ftp.FTPFileSystem.getFileStatus(FTPFileSystem.java:390)
    at org.apache.hadoop.fs.FileSystem.globStatusInternal(FileSystem.java:1701)
    at org.apache.hadoop.fs.FileSystem.globStatus(FileSystem.java:1647)
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:222)
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:270)
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
    at org.apache.spark.api.python.PythonRDD.getPartitions(PythonRDD.scala:58)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1929)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:927)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
    at org.apache.spark.rdd.RDD.collect(RDD.scala:926)
    at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)
    at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
    at py4j.Gateway.invoke(Gateway.java:259)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:209)
    at java.lang.Thread.run(Thread.java:745)

它说在端口 21 上的 192.168.125.124 上登录失败。似乎它在将结果返回到 ftp 时遇到了一些问题？我不明白结果是如何返回的。我从从属计算机的浏览器访问我的 ftp 服务器，它不提示登录。我的 vsftpd.conf 设置了以下权限。

anonymous_enable=YES
local_enable=YES
write_enable=YES
anon_upload_enable=YES
anon_mkdir_write_enable=YES
dirmessage_enable=YES
xferlog_enable=YES
listen=YES
no_anon_password=YES
anon_root=/srv/ftp

当我打破 map 并将部分简化为两个不同的语句时

>>> ff= sc.textFile("ftp://192.168.125.124/new.txt")
>>> df = ff.map(lambda s: len(s))
>>> df.reduce(lambda a, b: a + b)

map 部分运行良好，但我在减少时遇到相同的错误。我在我的集​​群上成功运行了其他不需要外部数据集的正常作业。例如

>>> data=[f for x in xrange(10000)]
>>> distData=sc.parallelize(data);
>>>distData.reduce(lambda a, b: a + b)

现在请告诉我如何解决这个问题以及我做错了什么。我可以使用 hdfs，但我想知道使用 ftp 有什么问题。

Answer 1

看看this :

你的 url 应该是 ftp://username:password@host/file 的形式。对于匿名登录，使用anonymous作为用户和密码可以是任何东西，它不能为空。

ff= sc.textFile("ftp://anonymous:pandamagic@192.168.125.124/new.txt")

而您只在 reduce 部分看到错误的原因是因为 map 是一个转换而 reduce 是一个 Action 。仅当对其调用操作时，RDD 才会具体化。