从分区的镶木地板文件中读取 DataFrame

Question

如何以数据帧的条件读取分区镶木地板，

这很好用，

val dataframe = sqlContext.read.parquet("file:///home/msoproj/dev_data/dev_output/aln/partitions/data=jDD/year=2015/month=10/day=25/*")

有 day=1 to day=30 的分区是否可以读取类似 (day = 5 to 6) 或 day=5,day=6、

的内容

val dataframe = sqlContext.read.parquet("file:///home/msoproj/dev_data/dev_output/aln/partitions/data=jDD/year=2015/month=10/day=??/*")

如果我输入 * 它会给我所有 30 天的数据而且它太大了。

Answer 1

sqlContext.read.parquet可以取多个路径作为输入。如果您只需要 day=5 和 day=6，您可以简单地添加两个路径，例如：

val dataframe = sqlContext
      .read.parquet("file:///your/path/data=jDD/year=2015/month=10/day=5/", 
                    "file:///your/path/data=jDD/year=2015/month=10/day=6/")

如果您在 day=X 下有文件夹，例如 country=XX，country 将自动添加为 dataframe 中的一列。

EDIT: As of Spark 1.6 one needs to provide a "basepath"-option in order for Spark to generate columns automatically. In Spark 1.6.x the above would have to be re-written like this to create a dataframe with the columns "data", "year", "month" and "day":

val dataframe = sqlContext
     .read
     .option("basePath", "file:///your/path/")
     .parquet("file:///your/path/data=jDD/year=2015/month=10/day=5/", 
                    "file:///your/path/data=jDD/year=2015/month=10/day=6/")

Answer 2

您需要提供mergeSchema = true选项。如下所述（来自 1.6.0）：

val dataframe = sqlContext.read.option("mergeSchema", "true").parquet("file:///your/path/data=jDD")

这会将所有 parquet 文件读入数据框，并在数据框数据中创建年、月和日列。

参考：https://spark.apache.org/docs/1.6.0/sql-programming-guide.html#schema-merging

Answer 3

如果你想读取多天，例如 day = 5 和 day = 6 并且想在路径本身中提及范围，可以使用通配符：

val dataframe = sqlContext
  .read
  .parquet("file:///your/path/data=jDD/year=2015/month=10/day={5,6}/*")

通配符也可用于指定天数范围：

val dataframe = sqlContext
  .read
  .parquet("file:///your/path/data=jDD/year=2015/month=10/day=[5-10]/*")

这匹配从 5 到 10 的所有日期。

Answer 4

在我的 pyspark 案例中：

sdf_table = spark.read.parquet("s3://bucket/table/**/*.parquet")

**是parquet的所有分区（一个glob表达式）

请注意，读取存储桶“table/”中的所有文件镶木地板，因此请注意其他文件

Answer 5

我无法像 Cristian 建议的那样使用通配符。经过一番挖掘，我发现以下似乎有效：

spark.read.option("recursiveFileLookup", "true").parquet("s3a://bucket/table/")

https://kontext.tech/article/887/read-parquet-files-from-nested-directories

从分区的镶木地板文件中读取 DataFrame

Reading DataFrame from partitioned parquet file

scala

apache-spark

parquet

spark-dataframe