在 pyspark 中的简单 .first()、.count() 中出现错误，

Posted 2023-04-15

【中文标题】在 pyspark 中的简单 .first()、.count() 中出现错误，

【英文标题】：getting error in simple .first(), .count() in pyspark,

【发布时间】：2019-12-23 14:36:47

【问题描述】：

我在 windows10 jupyter notebook 上运行以下代码，

airline= 'C:\spark-3.0.0-preview-bin-hadoop2.7\airlines.csv'
airlines = sc.textFile(airline)
print(airlines)

C:\spark-3.0.0-preview-bin-hadoop2.7irlines.csv MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:0

（这给了我正确的输出）

当我尝试运行print(airlines.first()) 我收到错误：

Py4JJavaError                             Traceback (most recent call last)
<ipython-input-13-36bc0976264b> in <module>
----> 1 print(airlines.first())
.
.
.
Py4JJavaError: An error occurred while calling o30.partitions.
: org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:/C:/spark-3.0.0-preview-bin-hadoop2.7irlines.csv
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:287)
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:229)
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:315)
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:205)
    at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:276)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:272)
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:49)
    at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:276)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:272)
    at org.apache.spark.api.java.JavaRDDLike.partitions(JavaRDDLike.scala:61)
    at org.apache.spark.api.java.JavaRDDLike.partitions$(JavaRDDLike.scala:61)
    at org.apache.spark.api.java.AbstractJavaRDDLike.partitions(JavaRDDLike.scala:45)
    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:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)

【问题讨论】：

这里的sc 是什么？

Spark 很懒惰。这意味着在需要文件之前不会加载文件。因此，您的第一条语句不会导致任何错误，它不会加载文件。一旦您调用collect、first、toPandas、show 或任何其他“Action”，将执行所需的“转换”，例如sc.textFile。该错误只是说明您的文件在给定的输入路径上不存在。看起来很容易解决！

@akshat, sc 是 spark 上下文对象的默认值

【参考方案1】：

我认为您提供的输入路径是正确的。如果没有，您必须对其进行审查。进一步在您的代码中进行以下更正：

代替：

airline= 'C:\spark-3.0.0-preview-bin-hadoop2.7\airlines.csv'

如下写：

airline= 'C:\\spark-3.0.0-preview-bin-hadoop2.7\\airlines.csv'

（或）

airline= 'C:/spark-3.0.0-preview-bin-hadoop2.7/airlines.csv'

代替：

print(airlines)

如下写：

   for i in airline.take(10): print (i)

（或）

   for i in airline.collect(): print (i)

注意：仅当您的文件较小时才使用 collect()，否则使用 take(n)，其中 n 是整数。

【讨论】：

谢谢Sri_Karthik，制作（双\）帮助了我

【参考方案2】：

我无法发表评论，所以我必须使用答案提出这个简单的问题：

你确定你写在变量中的路径是正确的吗？看消息似乎不是这样：

..hadoop2.7irlines.csv /ai" 不见了？