基本需求

將Keyed RDD[(Key,Value)]按Key儲存到不同檔案。

測試資料

資料格式：id,studentId,language,math,english,classId,departmentId

1,111,68,69,90,Class1,Economy
2,112,73,80,96,Class1,Economy
3,113,90,74,75,Class1,Economy
4,114,89,94,93,Class1,Economy
5,115,99,93,89,Class1,Economy
6,121,96,74,79,Class2,Economy
7,122,89,86,85,Class2,Economy
8,123,70,78,61,Class2,Economy
9,124,76,70,76,Class2,Economy
10,211,89,93,60,Class1,English
11,212,76,83,75,Class1,English
12,213,71,94,90,Class1,English
13,214,94,94,66,Class1,English
14,215,84,82,73,Class1,English
15,216,85,74,93,Class1,English
16,221,77,99,61,Class2,English
17,222,80,78,96,Class2,English
18,223,79,74,96,Class2,English
19,224,75,80,78,Class2,English
20,225,82,85,63,Class2,English
 

用Spark RDD實現

package com.bigData.spark

import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
import org.apache.log4j.{Level, Logger}
import org.apache.spark.HashPartitioner
import org.apache.spark.sql.SparkSession

/**
  * Author: Wang Pei
  * License: Copyright(c) Pei.Wang
  * Summary:
  *   RDD 按Key儲存到不同檔案
  */
 

object OutputToMultiFile {
  def main(args: Array[String]): Unit = {


    /**設定日誌等級*/
    Logger.getLogger("org").setLevel(Level.WARN)

    /**spark環境*/
    val spark = SparkSession.builder().master("local[3]").appName(this.getClass.getSimpleName.replace("$","")).getOrCreate()

    /**Keyed RDD*/
    val data =
 
spark.sparkContext.textFile("data/scores.csv")
        //Keyed RDD
        .map(item=>(item.split(",").takeRight(2).reverse.mkString("_"),item))
        //按Key Hash分割槽，4個Key分到4個Partition中
        .partitionBy(new HashPartitioner(4))


    /**按Key儲存到不同檔案*/
    data.saveAsHadoopFile("data/multiKeyedDir",
      classOf[String],
      classOf[String],
      classOf[PairRDDMultipleTextOutputFormat]
    )

    spark.stop()

  }

}

/**繼承類重寫方法*/
class PairRDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any, Any] {
  //1)檔名：根據key和value自定義輸出檔名。
  override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String ={
    val fileNamePrefix=key.asInstanceOf[String]
    val fileName=fileNamePrefix+"-"+name
    fileName
  }

  //2)檔案內容：默認同時輸出key和value。這裡指定不輸出key。
  override def generateActualKey(key: Any, value: Any): String = {
    null
  }
}