scala – 通过Spark阅读保存在文件夹中的所有Parquet文件

发布时间：2020-12-16 19:19:29 所属栏目：安全来源：网络整理

导读：我有一个包含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))

我有一个包含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文件夹中的所有镶木地板文件时保存数据帧后,它给了我错误.

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文件,但如果有办法读取文件夹中的所有镶木地板文件会更好.

解决方法

Spark不会以您认为的方式写入/读取镶木地板.

它使用Hadoop库来编写/读取分区的镶木地板文件.

因此,您的第一个镶木地板文件位于路径/tmp/test/df/1.parquet/下,其中1.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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!