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

Question

我们正在运行spark 2.3.0 AWW EMR.以下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

所以没有写木地板文件.

我已经尝试过这种情况,可能是二十次,两次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

Answer 1

这不是一个错误,它是预期的行为.Spark并不真正支持对非分布式存储的写入(它将在local模式下工作,因为您有共享文件系统).

本地路径不会(仅)解释为驱动程序上的路径(这需要收集数据),而是解释每个执行程序上的本地路径.因此,每个执行程序都会将自己的块写入其自己的本地文件系统.

不仅输出不可读(为每个执行器加载数据而驱动程序应该看到文件系统的相同状态),但是根据提交算法,可能甚至不能完成(从临时目录移动).