我想将一个大的镶木地板文件拆分成HDFS中不同文件夹中的多个镶木地板文件，这样我就可以在其上构建分区表（无论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)
}
}
刚刚抛出一个零点异常 .
1 回答
您应该使用DataFrame中的partitionBy . 你不需要groupBy . 像下面这样的东西应该给你想要的东西 .