Skip to main content
 首页 » 编程设计

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

2024年08月22日44zhujiabin

嘿，我是 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 有什么问题。

请您参考如下方法:

看看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 才会具体化。


  文章来源：
  https://blog.windname.com/post/gfhad2b7.html
阅读延展