scala - 通过 Spark 读取文件夹中保存的所有 Parquet 文件

Question

我有一个包含 Parquet 文件的文件夹。像这样的事情:

scala> val df = sc.parallelize(List(1,2,3,4)).toDF()
df: org.apache.spark.sql.DataFrame = [value: int]

scala> df.write.parquet("/tmp/test/df/1.parquet")

scala> val df = sc.parallelize(List(5,6,7,8)).toDF()
df: org.apache.spark.sql.DataFrame = [value: int]

scala> df.write.parquet("/tmp/test/df/2.parquet")

保存数据帧后，当我去读取 df 文件夹中的所有 Parquet 文件时，出现错误。

scala> val read = spark.read.parquet("/tmp/test/df")
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$8.apply(DataSource.scala:189)
  at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$8.apply(DataSource.scala:189)
  at scala.Option.getOrElse(Option.scala:121)
  at org.apache.spark.sql.execution.datasources.DataSource.org$apache$spark$sql$execution$datasources$DataSource$$getOrInferFileFormatSchema(DataSource.scala:188)
  at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:387)
  at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:152)
  at org.apache.spark.sql.DataFrameReader.parquet(DataFrameReader.scala:441)
  at org.apache.spark.sql.DataFrameReader.parquet(DataFrameReader.scala:425)
  ... 48 elided

我知道我可以通过提供完整路径来读取 Parquet 文件，但如果有一种方法可以读取文件夹中的所有 Parquet 文件，那就更好了。

Answer 1

Spark 不会按照您想象的方式写入/读取 parquet。

它使用 Hadoop 库来写入/读取分区 parquet 文件。

因此，您的第一个 parquet 文件位于路径 /tmp/test/df/1.parquet/ 下，其中 1.parquet 是一个目录。这意味着当从 parquet 读取数据时，您需要提供 parquet 目录的路径或路径(如果它是一个文件)。

val df = spark.read.parquet("/tmp/test/df/1.parquet/")

我建议您阅读官方文档以了解更多详细信息。 [参见。 SQL Programming Guide - Parquet Files ]

编辑:

您一定正在寻找这样的东西:

scala> sqlContext.range(1,100).write.save("/tmp/test/df/1.parquet")

scala> sqlContext.range(100,500).write.save("/tmp/test/df/2.parquet")

scala> val df = sqlContext.read.load("/tmp/test/df/*")
// df: org.apache.spark.sql.DataFrame = [id: bigint]

scala> df.show(3)
// +---+
// | id|
// +---+
// |400|
// |401|
// |402|
// +---+
// only showing top 3 rows

scala> df.count
// res3: Long = 499

您还可以在文件路径 URI 中使用通配符。

您可以提供多个文件路径，如下所示:

scala> val df2 = sqlContext.read.load("/tmp/test/df/1.parquet","/tmp/test/df/2.parquet")
// df2: org.apache.spark.sql.DataFrame = [id: bigint]

scala> df2.count
// res5: Long = 499