通过一个键将一个大 parquet 文件拆分为多个 parquet 文件

Question

我想将一个大 parquet 文件拆分为 HDFS 中不同文件夹中的多个 parquet 文件，以便我可以在其上构建分区表（无论 Hive/Drill/Spark SQL）。

数据示例：

+-----+------+
|model|  num1|
+-----+------+
|  V80| 195.0|
|  V80| 750.0|
|  V80| 101.0|
|  V80|   0.0|
|  V80|   0.0|
|  V80| 720.0|
|  V80|1360.0|
|  V80| 162.0|
|  V80| 150.0|
|  V90| 450.0|
|  V90| 189.0|
|  V90| 400.0|
|  V90| 120.0|
|  V90|  20.3|
|  V90|   0.0|
|  V90|  84.0|
|  V90| 555.0|
|  V90|   0.0|
|  V90|   9.0|
|  V90|  75.6|
+-----+------+

结果文件夹结构应按“模型”字段分组：

+
|
+-----model=V80
|       | 
|       +----- XXX.parquet
+-----model=V90
|       | 
|       +----- XXX.parquet

我试过这样的脚本：

def main(args: Array[String]): Unit = {
   val conf = new SparkConf()
   case class Infos(name:String, name1:String)
    val sc = new SparkContext(conf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val rdd = sqlContext.read.load("hdfs://nameservice1/user/hive/warehouse/a_e550_parquet").select("model", "num1").limit(10000)

    val tmpRDD = rdd.map { item => (item(0), Infos(item.getString(0), item.getString(1))) }.groupByKey()

    for (item <- tmpRDD) {
      import sqlContext.implicits._
      val df = item._2.toSeq.toDF()
      df.write.mode(SaveMode.Overwrite).parquet("hdfs://nameservice1/tmp/model=" + item._1)
    }
  }

刚刚抛出了一个空点异常。

Answer 1

您应该使用 DataFrame 中的 partitionBy。您不需要 groupBy。像下面这样的东西应该可以满足您的需求。

val df = sqlContext.read.parquet("hdfs://nameservice1/user/hive/warehouse/a_e550_parquet").select("model", "num1").limit(10000)
df.write.partitionBy("model").mode(SaveMode.Overwrite)

Answer 2

如果我不想要 partitionBy... 只需将文件按列拆分，使用某种边界，例如 分区 1，使用 column1，值从 1 到 10000 分区 2，使用 column1，值从 10001 到 20000 分区 3，使用 column1，值从 20001 到 30000 .....

等等！