spark多文件輸出實例

實例需求

統計每個月每個家庭成員給自己打電話的次數，並按月份輸出到不同文件中

數據集

數據集下載點擊這里

數據格式：編號 聯系人 電話號碼 日期

思路分析

map階段：將 聯系人 + 電話號碼 + 日期中的月份 組合成的作為 key，value為1
reduce階段：相同key做value相加操作，然后利用hadoop的MultipleTextOutputFormat，實現按月份將不同月份的統計數據輸出到不同的文件中

實踐步驟

1.讀取數據文件，生成RDD

val input = "/data/spark-example/phone/phone.txt"
val fileRDD = sc.textFile(input)

2.將 聯系人 + 電話號碼 + 日期中的月份 組合成的作為 key，value為1

val mapRDD = fileRDD.map(line => (line.split("\t")(1) + "\t" + line.split("\t")(3).split("-")(1) + "\t" + line.split("\t")(2), 1))

3.相同key做value相加操作

val reduceRDD = mapRDD.reduceByKey(_ + _)

4.自定義多文件輸出格式類，繼承MultipleTextOutputFormat類，重寫generateFileNameForKeyValue方法

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any, Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
        key.asInstanceOf[String].split("\t")(1) + ".txt"
    }
}

5.輸出統計結果

reduceRDD.saveAsHadoopFile("hdfs://oldsix1:9000/data/spark-example/phone/output/", classOf[String], classOf[String], classOf[RDDMultipleTextOutputFormat])

完整代碼

1.RDDMultipleTextOutputFormat.scala

import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any, Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
        key.asInstanceOf[String].split("\t")(1) + ".txt"
    }
}

2.PhoneCount.scala

import org.apache.spark.{SparkConf, SparkContext}

object PhoneCount {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("phone-count")
        val sc = new SparkContext(conf)
        val input = "/data/spark-example/phone/phone.txt"
        val fileRDD = sc.textFile(input)
        val mapRDD = fileRDD.map(line => (line.split("\t")(1) + "\t" + line.split("\t")(3).split("-")(1) + "\t" + line.split("\t")(2), 1))
        val reduceRDD = mapRDD.reduceByKey(_ + _)
        reduceRDD.saveAsHadoopFile("hdfs://oldsix1:9000/data/spark-example/phone/output/", classOf[String], classOf[String], classOf[RDDMultipleTextOutputFormat])
    }
}

結果查看