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])
    }
}

结果查看