在火花数据帧上应用操作时出错答案

【问题标题】：error in applying actions on spark dataframe在火花数据帧上应用操作时出错
【发布时间】：2017-05-24 00:55:24
【问题描述】：

我是 Spark 框架的新手，并且在我的本地机器上处理一些 (~) 小任务来练习。我的任务如下：我在 S3 中存储了 365 个压缩的 csv 文件，其中包含每日日志。我想构建一个全年的数据框。我的方法是从存储桶中检索密钥，构建每日数据帧，将它们统一为月份数据帧，为它们做同样的事情，并获得一整年的数据帧作为回报。

这对我检索到的一些样本数据进行了测试。在构建 DataFrame 之前，我解压缩了文件，将未压缩的 csv 文件写入磁盘，并使用它来创建 DataFrame。

问题：如果我从磁盘中删除 csv 文件（将其设为临时文件），则在创建数据框后，我无法对数据框执行任何操作（例如 year_df.count()）。抛出 Spark.exception：

“作业因阶段失败而中止：.... java.io.FileNotFoundException: 文件 .csv 不存在”

在对 SO 进行一些搜索后，我发现原因可能是 Spark 在对 DataFrame 应用 SQL 查询时使用的 MetaData (External Table not getting updated from parquet files written by spark streaming)。我改了

spark.sql.parquet.cache元数据

通过运行spark = SparkSession.builder.config("spark.sql.parquet.cacheMetadata", "false").getOrCreate()。确保 spark.conf.get("spark.sql.parquet.cacheMetadata") 返回 false..

找不到任何解决方案。当然，将所有文件解压缩到 S3 中也可以，但这对我没有用..

谢谢！

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

Spark 以惰性方式执行操作。

这意味着，如果您可以进行少量转换，但文件只会在您调用操作时被读取。

它在数据集上的工作方式与在 RDD 中的工作方式相同，因为数据集由 RDD 支持

考虑代码：

val df = sqlContext.read // read file
val query = df.groupBy('x).count()

query.show() // here the data will be read

因此，如果您在读取文件之前将其删除，则 Spark 将引发异常。您可以通过执行一些操作来强制阅读，即take 或show。如果你会这样做，它将被缓存cache():

val df = sqlContext.read // read file
val query = df.groupBy('x).count().cache()

query.show() // here the data will be read and cached

【讨论】：

您好，感谢您的回复。因此，如果我理解正确，只有在我缓存了我感兴趣的查询并调用了一个操作来计算它们之后，我才能安全地删除源文件（在这种情况下是我的 .csv）。所以，我有 2 个选择：要么将未压缩的 CSV 文件保存到我的 S3 以进行持续访问，要么直接读取 gzip 压缩文件以触发，并在一些转换后创建 DataFrame。对吗？
@Mike Flow: 读取 -> 缓存 -> 一些操作 -> 删除文件应该没问题 - 除非你发现一些不寻常的 Spark 行为；）你是对的，我会将文件保存在 S3 中，直到处理完成done - 万一出现故障，您将有机会重新处理文件

【解决方案2】：

Cache() 仍然只是一个提示； Spark 可能需要在出现故障时重新计算值，或者只是由于缓存压力而丢弃缓存的数据。如果您想删除源数据，请确保您已将结果写出，并且真的不再需要这些数据了。

我实际上建议将 CSV 转换为任何列格式（ORC、Parquet）并使用 Snappy 进行压缩。处理效率更高，尤其是谓词下推

【讨论】：