使用 SQlcontext 在 spark 中加载我的 csv 时出现错误答案

【问题标题】：I am getting error while loading my csv in spark using SQlcontext使用 SQlcontext 在 spark 中加载我的 csv 时出现错误
【发布时间】：2018-08-24 11:46:06
【问题描述】：

enter image description here

sparkDF = sqlContext.read.format('com.databricks.spark.csv').options(header='true').load('PR_DATA_35.csv')

错误_ -------------------------------------------------- -------------------------------------------------- ------------------------------------------------ Py4JJavaError Traceback （最近一次通话最后）在（） ----> 1 sparkDF = sqlContext.read.format('com.databricks.spark.csv') .options(header='true').load('PR_DATA_35.csv')

/home/ec2-user/spark/python/pyspark/sql/readwriter.pyc in load(self, 路径、格式、架构、**选项）第157章 158 如果是实例（路径，基本字符串）： --> 159 返回 self._df(self._jreader.load(path)) 160 elif 路径不是无： 161如果类型（路径）！=列表：

/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：调用 o312.load 时出错。： org.apache.spark.SparkException：作业因阶段失败而中止：阶段 5.0 中的任务 0 失败 4 次，最近一次失败：丢失任务 0.3 在阶段 5.0 (TID 23, 172.31.17.233, executor 0)： java.io.FileNotFoundException：文件文件：/home/ec2-user/PR_DATA_35.csv 不存在可能基础文件已更新。您可以通过运行“REFRESH”显式地使 Spark 中的缓存无效 SQL 中的 TABLE tableName' 命令或通过重新创建 Dataset/DataFrame 涉及。在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.org$apache$spark$sql$execution$datasources$FileScanRDD$$anon$$readCurrentFile(FileScanRDD.scala:127) 在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:174) 在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:105) 在 org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext（未知来源）在 org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) 在 org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395) 在 org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234) 在 org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) 在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:108) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 在 java.lang.Thread.run(Thread.java:748)

驱动程序堆栈跟踪：在 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1499) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1487) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1486) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1486) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814) 在 scala.Option.foreach(Option.scala:257) 在 org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:814) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1714) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1669) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1658) 在 org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:630) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:2022) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:2043) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:2062) 在 org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:336) 在 org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38) 在 org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:2853) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2153) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2153) 在 org.apache.spark.sql.Dataset$$anonfun$55.apply(Dataset.scala:2837) 在 org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65) 在 org.apache.spark.sql.Dataset.withAction(Dataset.scala:2836) 在 org.apache.spark.sql.Dataset.head(Dataset.scala:2153) 在 org.apache.spark.sql.Dataset.take(Dataset.scala:2366) 在 org.apache.spark.sql.execution.datasources.csv.TextInputCSVDataSource$.infer(CSVDataSource.scala:147) 在 org.apache.spark.sql.execution.datasources.csv.CSVDataSource.inferSchema(CSVDataSource.scala:62) 在 org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.inferSchema(CSVFileFormat.scala:57) 在 org.apache.spark.sql.execution.datasources.DataSource$$anonfun$7.apply(DataSource.scala:177) 在 org.apache.spark.sql.execution.datasources.DataSource$$anonfun$7.apply(DataSource.scala:177) 在 scala.Option.orElse(Option.scala:289) 在 org.apache.spark.sql.execution.datasources.DataSource.getOrInferFileFormatSchema(DataSource.scala:176) 在 org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:366) 在 org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:178) 在 org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:156) 在 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) 原因： java.io.FileNotFoundException：文件文件：/home/ec2-user/PR_DATA_35.csv 不存在可能基础文件已更新。您可以通过运行“REFRESH”显式地使 Spark 中的缓存无效 SQL 中的 TABLE tableName' 命令或通过重新创建 Dataset/DataFrame 涉及。在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.org$apache$spark$sql$execution$datasources$FileScanRDD$$anon$$readCurrentFile(FileScanRDD.scala:127) 在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:174) 在 org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:105) 在 org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext（未知来源）在 org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) 在 org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395) 在 org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234) 在 org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827) 在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:108) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ... 1 更多

感谢任何帮助。谢谢

【问题讨论】：

这是一个FileNotFound 异常，'/home/ec2- user/PR_DATA_35.csv' 是否正确？路径中有多余的空格，是拼写错误还是？
不，它只是在堆栈溢出时编辑时发生了变化。
您是否为com.databricks.spark.csv 包含了相应的jar（即使这会引发ClassNotFound 异常）？你说它适用于熊猫，我假设在本地，火花代码在哪里运行？也是本地的吗？
在 ec2 上它适用于 pandas，因为我使用 aws cli 将数据从 s3 提取到我的 was 实例。
你检查过 pandas 使用的路径吗？我猜 pandas 正在从 ec2 路径读取，但 spark 代码正在尝试从本地读取。

标签： python amazon-web-services apache-spark pyspark databricks

【解决方案1】：

原因：java.io.FileNotFoundException：文件文件：/home/ec2-user/PR_DATA_35.csv 不存在？？

【讨论】：

我使用 pandas 进行了交叉检查。它适用于 pandas，但不适用于 spark。

【解决方案2】：

您的代码没有找到要转换为 DataFrame 的文件

Py4JJavaError: An error occurred while calling o312.load.: 
Job aborted due to stage failure: Task 0 in stage 5.0 failed 4 times,
most recent failure: Lost task 0.3 in stage 5.0 (TID 23, 172.31.17.233,
executor 0): java.io.FileNotFoundException: File file:/home/ec2
user/PR_DATA_35.csv does not exist It is possible the underlying files
have been updated.

我有一些建议

验证文件是否存储在您的 EC2 实例中，并且它与您的 spark 代码位于同一文件夹中。
按照 Databricks guide，您可以指定文件的完整路径，对于 ec2 实例，它应该类似于：/home/ubuntu/project_folder/csv_file.csv
你可以看看这个stackoverflow问题

这是我读取 csv 文件的方式

# submit to spark on my dev env
# ./bin/spark-submit /Users/estebance/Documents/Projects/tests/spark_csv.py
# Spark version 2.0 and up, spark-csv is part of core Spark functionality and doesn't require a separate library. So you could just do for example
from pyspark import SparkContext
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName("CsvReader").getOrCreate()
csv_df = spark.read.format("csv").option("header", "true").load("/Users/estebance/Documents/Projects/tests/sample.csv")
print(csv_df.head(2))

希望对你有帮助，最好的问候

【讨论】：

1.我已经在同一个文件夹中检查了它。 2. 我对数据块做了同样的事情。 3.我试过内置读取csv文件，但错误是一样的。
我已经用我在开发环境中测试过的一些代码更新了答案，我对 spark 了解不多，但如果你仍然遇到错误，你应该看看到安装

【解决方案3】：

与其他人所说的相反，此错误通常发生在 Spark 中，当您从文件夹读取数据、修改它并保存在您最初读取的数据之上时。

只需尝试将第一组数据保存在临时文件夹中或使用answer 中描述的解决方案。

【讨论】：