在 Pyspark 中动态读取 Hive 表分区文件

Question

我在Pyspark 中有一个变量，如下所示

range = (
        (load_date='2020-06-10' AND critical in ('2', '3', '4') OR 
        (load_date='2020-06-11' AND critical in ('1', '2', '3', '4') OR
        (load_date='2020-06-12' AND critical in ('1', '2', '3', '4') OR 
        (load_date='2020-06-13' AND critical in ('1', '2', '3') 
        )

现在我想读取格式如下的hive 表分区数据文件

/user/$USER/load_date=2020-06-10/critical=2/some_files
/user/$USER/load_date=2020-06-10/critical=3/some_files
/user/$USER/load_date=2020-06-10/critical=4/some_files
/user/$USER/load_date=2020-06-11/critical=1/some_files
/user/$USER/load_date=2020-06-11/critical=2/some_files
/user/$USER/load_date=2020-06-11/critical=3/some_files
/user/$USER/load_date=2020-06-11/critical=4/some_files
/user/$USER/load_date=2020-06-12/critical=1/some_files
/user/$USER/load_date=2020-06-12/critical=2/some_files
/user/$USER/load_date=2020-06-12/critical=3/some_files
/user/$USER/load_date=2020-06-12/critical=4/some_files
/user/$USER/load_date=2020-06-13/critical=1/some_files
/user/$USER/load_date=2020-06-13/critical=2/some_files
/user/$USER/load_date=2020-06-13/critical=3/some_files

由于我只能访问 hive 表的 hdfs 路径而不是实际表，因此我无法直接从 hive 表创建数据框

现在使用这些文件，我想创建一个如下所示的数据框。

df = spark.read
        .option("basePath", "/user/$USER")
        .parquet("/user/$USER/load_date=2020-06-10/critical=2/some_files", 
                "/user/$USER/load_date=2020-06-10/critical=3/some_files", 
                "/user/$USER/load_date=2020-06-10/critical=4/some_files", 
                "/user/$USER/load_date=2020-06-11/critical=1/some_files", 
                "/user/$USER/load_date=2020-06-11/critical=2/some_files", 
                "/user/$USER/load_date=2020-06-11/critical=3/some_files", 
                "/user/$USER/load_date=2020-06-11/critical=4/some_files", 
                "/user/$USER/load_date=2020-06-12/critical=1/some_files", 
                "/user/$USER/load_date=2020-06-12/critical=2/some_files", 
                "/user/$USER/load_date=2020-06-12/critical=3/some_files", 
                "/user/$USER/load_date=2020-06-12/critical=4/some_files", 
                "/user/$USER/load_date=2020-06-13/critical=1/some_files", 
                "/user/$USER/load_date=2020-06-13/critical=2/some_files", 
                "/user/$USER/load_date=2020-06-13/critical=3/some_files")

我能够创建数据框，但是以这种方式传递所有文件路径是一种乏味且不是最佳方式的方式。我想以更简单的动态方式来做到这一点

我已经尝试过如下

df = spark.read
        .option("basePath", "/user/$USER")
        .parquet("/user/$USER/load_date=*/critical=*")

上述方法也适用并创建数据框。

但对我来说问题是，如果我使用上述方法，那么我可能会有更多不需要的数据。

例如如果我有一个文件路径

"/user/$USER/load_date=2020-06-13/critical=4/some_files"

我暂时不想读取上述路径中的文件。

requirement

使用range 变量提取load_date 和critical 值并仅读取这些文件路径

我怎样才能做到这一点？

Answer 1

您可以使用spark.read.parquet(<root-dir>) 读取根目录，然后应用where 子句。这会将 wehreclause 中的谓词推送到源路径。

让我们通过一个例子来理解这一点- 试试这个-

1。创建由batch_id 分区的文件

这就是我的情况/Users/sokale/models/run_1/batch_id=73/part-00001-5fa5aebb-a836-43d2-97d2-7cf9bb722c26.c000.snappy.parquet

  val df = spark.range(1,5)
      .withColumn("batch_id", lit(70) + col("id"))

    df.show(false)
    df.printSchema()

    /**
      * +---+--------+
      * |id |batch_id|
      * +---+--------+
      * |1  |71      |
      * |2  |72      |
      * |3  |73      |
      * |4  |74      |
      * +---+--------+
      *
      * root
      * |-- id: long (nullable = false)
      * |-- batch_id: long (nullable = false)
      */

    df.write.partitionBy("batch_id")
      .mode(SaveMode.Overwrite)
      .parquet("/Users/sokale/models/run_1")
    /**
      * $ cd run_1/
      * $ ls -l
      * total 0
      * ............ _SUCCESS
      * ............ batch_id=71
      * ............ batch_id=72
      * ............ batch_id=73
      * ............ batch_id=74
      */

2。读取分区 barch_id=73 的 parquet 文件

来自spark doc-

spark.sql.parquet.filterPushdown    default-true    Enables Parquet filter push-down optimization when set to true.

这意味着df.read.parquet(dir).where(partitionCOndition)只使用过滤器下推读取指定的分区目录

 // read only file with batch_id=73
    spark.read.parquet("/Users/sokale/models/run_1").where(col("batch_id").equalTo(73))
      .show(false)

    /**
      * +---+--------+
      * |id |batch_id|
      * +---+--------+
      * |3  |73      |
      * +---+--------+
      */
    // read all partitions
    val readDF = spark.read.parquet("/Users/sokale/models/run_1")
    readDF.show(false)
    readDF.printSchema()

    /**
      * +---+--------+
      * |id |batch_id|
      * +---+--------+
      * |3  |73      |
      * |2  |72      |
      * |1  |71      |
      * |4  |74      |
      * +---+--------+
      *
      * root
      * |-- id: long (nullable = true)
      * |-- batch_id: integer (nullable = true)
      */