專案中需要將spark的輸出按id輸出到不同的目錄中，即實現在spark中的多路輸出。我們可以呼叫saveAsHadoopFile函式並自定義一個OutputFormat類，就可以達到上述目的。

import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by WangLei on 17-6-2.
  */
 

object genSpecifiedUserData {

    val inputPath = "xxx"

    def genData(sc:SparkContext) = {
        sc.textFile(inputPath)
            .filter(x => StringUtils.split(x, "\t").length == 7)
            .map(x => {
                val lines = StringUtils.split(x, "\t")
                val (id, aid, title, desc) = (lines(0
 
), lines(1), lines(5), lines(6))
                (id, aid, title, desc)
            })
            .filter(x => StringUtils.isNotBlank(x._1))
            .map(x => {
                val (id, aid, title, desc) = (x._1, x._2, x._3, x._4)
                val pattern = Array(aid, title, desc).mkString("\t"
 
)
                (id, pattern)
            })
            .partitionBy(new org.apache.spark.HashPartitioner(10))
    }

    def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName("gen_specified_user_data")
        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        val sc = new SparkContext(sparkConf)

        val output = "/data/xxx/specified/"
        val fileSystem = FileSystem.get(sc.hadoopConfiguration)
        fileSystem.delete(new Path(output), true)
        findSpecifiedUser(sc).saveAsHadoopFile(
            output,
            classOf[String],
            classOf[String],
            classOf[RDDMultipleTextOutputFormat[_, _]])

        sc.stop()
    }
}

class RDDMultipleTextOutputFormat[K, V]() extends MultipleTextOutputFormat[K, V]() {
    override def generateFileNameForKeyValue(key: K, value: V, name: String) : String = {
        (key + "/" + name)
    }
}

其中，輸入的資料是以”\t”分隔一共七列，第一列為使用者id。我們希望將輸出的時候，相同的使用者id輸出到同一個目錄下面，不同的使用者id分開。

RDDMultipleTextOutputFormat類中的generateFileNameForKeyValue函式有三個引數，key和value就是我們RDD的Key和Value，而name引數是每個Reduce的編號。上面的程式碼中沒有使用該引數，而是直接將同一個Key的資料輸出到同一個檔案中。

最後生成的結果在HDFS上為：

/data/xxx/specified/idA/...
/data/xxx/specified/idB/...
/data/xxx/specified/idC/...