将数据框保存到本地文件系统会导致空结果

Posted 2023-03-28

【中文标题】将数据框保存到本地文件系统会导致空结果

【英文标题】：Saving dataframe to local file system results in empty results

【发布时间】：2019-01-07 06:16:08

【问题描述】：

我们正在AWS EMR 上运行 spark 2.3.0。以下DataFrame "df" 非空且大小适中：

scala> df.count
res0: Long = 4067

以下代码适用于将df 写入hdfs：

   scala> val hdf = spark.read.parquet("/tmp/topVendors")
hdf: org.apache.spark.sql.DataFrame = [displayName: string, cnt: bigint]

scala> hdf.count
res4: Long = 4067

但是，使用相同的代码写入本地 parquet 或 csv 文件会得到空结果：

df.repartition(1).write.mode("overwrite").parquet("file:///tmp/topVendors")

scala> val locdf = spark.read.parquet("file:///tmp/topVendors")
org.apache.spark.sql.AnalysisException: Unable to infer schema for Parquet. It must be specified manually.;
  at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$9.apply(DataSource.scala:207)
  at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$9.apply(DataSource.scala:207)
  at scala.Option.getOrElse(Option.scala:121)

我们可以看到它失败的原因：

 ls -l /tmp/topVendors
total 0
-rw-r--r-- 1 hadoop hadoop 0 Jul 30 22:38 _SUCCESS

所以没有拼花文件正在写入。

我已经尝试了大约 20 次，对于 csv 和 parquet 以及两个不同的 EMR 服务器：在所有情况下都表现出相同的行为。

这是 EMR 特定的错误吗？更一般的EC2 错误？还有什么？此代码适用于spark macOS。

以防万一 - 这是版本信息：

Release label:emr-5.13.0
Hadoop distribution:Amazon 2.8.3
Applications:Spark 2.3.0, Hive 2.3.2, Zeppelin 0.7.3

【问题讨论】：

只是好奇！！你是在运行执行程序的机器上做ls -l /tmp/topVendors 吗？

【参考方案1】：

这不是错误，而是预期的行为。 Spark 并不真正支持对非分布式存储的写入（它将在local 模式下工作，只是因为您拥有共享文件系统）。

本地路径不（仅）解释为驱动程序上的路径（这需要收集数据），而是每个执行程序上的本地路径。因此每个 executor 都会将自己的 chunk 写入自己的本地文件系统。

不仅输出不可读（加载数据时每个执行程序和驱动程序应该看到文件系统的相同状态），而且根据提交算法，甚至可能无法最终确定（从临时目录移动）。

【讨论】：

我明白了。有趣的是，我曾以为以前在集群上做过这个。也许它在 Standalone 中有效，但在 Yarn 中无效？？

我已经仔细检查过，在我们内部的非 AWS/EMR 集群上，我们可以写信给file://

AFAIK 似乎可行的情况是，驱动程序偶然与执行者位于同一地点。然而，这和本地一样，是资源分配的细节，而不是规则。

顺便说一句 Holden Karau - spark 上的***提交者有一个答案指出可以进行本地保存 ***.com/a/31240494/1056563 saveAsTextFile is able to take in local file system paths (e.g. file:///tmp/magic/...).

我在这里的答案与霍尔顿的答案之间没有真正的冲突。 Holden 明确指出，如果您在分布式集群上运行，您很可能希望 collect() 将数据返回到集群，然后使用标准文件操作将其保存，据我所知，不会'不声称能够接受本地文件系统路径意味着数据会自动传输到驱动程序。

【参考方案2】：

当您尝试将空目录读取为 parquet 时，通常会发生此错误。 你可以检查 1. 如果 DataFrame 为空，则在写入之前使用 output.rdd.isEmpty()。 2.检查你给的路径是否正确

您还以什么模式运行您的应用程序？如果您在集群模式下运行，请尝试在客户端模式下运行它。