如何在 pyspark AWS 上跨集群添加文件

Question

我是新来的火花。我正在尝试从我的主实例中读取一个文件，但我收到了这个错误。经过研究，我发现您需要将数据加载到 hdfs 或跨集群复制。我找不到执行其中任何一项的命令。

----------------------------------- ---------------------------- Py4JJavaError Traceback（最近调用 最后）在（） ----> 1 ncols = rdd.first().features.size # 数据集的列数（无类）

/home/ec2-user/spark/python/pyspark/rdd.pyc in first(self) 1359
ValueError: RDD 为空 1360 """ -> 1361 rs = self.take(1) 1362 if rs: 1363 return rs[0]

/home/ec2-user/spark/python/pyspark/rdd.pyc in take(self, num) 1311 """ 1312 项 = [] -> 1313 totalParts = self.getNumPartitions() 1314 partsScanned = 0 1315

/home/ec2-user/spark/python/pyspark/rdd.pyc in getNumPartitions(self) 2438 2439 def getNumPartitions（自我）： -> 2440 返回 self._prev_jrdd.partitions().size() 2441 2442 @property

/home/ec2-user/spark/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py 在 调用(self, *args) 1131 回答 = self.gateway_client.send_command(command) 1132 return_value = get_return_value( -> 1133 answer, self.gateway_client, self.target_id, self.name) 1134 1135 for temp_args in temp_args：

/home/ec2-user/spark/python/pyspark/sql/utils.pyc in deco(*a, **kw) 61 def deco(*a, **kw): 62 尝试： ---> 63 返回 f(*a, **kw) 64 除了 py4j.protocol.Py4JJavaError 作为 e: 65 秒 = e.java_exception.toString()

/home/ec2-user/spark/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py 在 get_return_value(answer, gateway_client, target_id, name) 第317章 318 “调用 {0}{1}{2} 时出错。\n”。 --> 319 格式（target_id，“.”，名称），值） 320 其他： 第321章

Py4JJavaError：调用 o122.partitions 时出错。 ： org.apache.hadoop.mapred.InvalidInputException：输入路径不 存在：文件：/home/ec2-user/PR_DATA_35.csv 在 org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:285) 在 org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:228) 在 org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313) 在 org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:194) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250) 在 scala.Option.getOrElse(Option.scala:121) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:250) 在 org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250) 在 scala.Option.getOrElse(Option.scala:121) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:250) 在 org.apache.spark.api.java.JavaRDDLike$class.partitions(JavaRDDLike.scala:61) 在 org.apache.spark.api.java.AbstractJavaRDDLike.partitions(JavaRDDLike.scala:45) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 在 py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 在 py4j.Gateway.invoke(Gateway.java:280) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection.run(GatewayConnection.java:214) 在 java.lang.Thread.run(Thread.java:748)

Answer 1

由于您已经在 AWS 中，因此将数据文件存储在 s3 中并直接从那里打开它们可能会更容易。