title: Split one big parquet file into multiple parquet files by a key
body: <p>I'd like split a big parquet file into multiple parquet files in different folder in HDFS, so that I can build partitioned table (whatever Hive/Drill/Spark SQL) on it. </p>
  
  <p>Data example:</p>
  
  <pre><code>+-----+------+
  |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|
  +-----+------+
  </code></pre>
  
  <p>The result folder structure should be grouped by "model" field:</p>
  
  <pre><code>+
  |
  +-----model=V80
  |       | 
  |       +----- XXX.parquet
  +-----model=V90
  |       | 
  |       +----- XXX.parquet
  </code></pre>
  
  <p>I tried the script like this:</p>
  
  <pre><code>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 =&gt; (item(0), Infos(item.getString(0), item.getString(1))) }.groupByKey()
  
      for (item &lt;- tmpRDD) {
        import sqlContext.implicits._
        val df = item._2.toSeq.toDF()
        df.write.mode(SaveMode.Overwrite).parquet("hdfs://nameservice1/tmp/model=" + item._1)
      }
    }
  </code></pre>
  
  <p>Just threw out a null point exception.</p>