Spark多级目录输出,且输出不同值,重写MultipleTextOutputFormat

60 阅读 0 评论 40 点赞

我是靠谱客的博主负责柠檬，最近开发中收集的这篇文章主要介绍Spark多级目录输出,且输出不同值,重写MultipleTextOutputFormat，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

数据结构

www.alibaba.com	222.38.194.133	440887	[09/20/2019 14:33:52 +0800]	河北	秦皇岛 铁通	2019	09	20
www.baidu.com	171.13.146.53	654665	[09/20/2019 14:33:52 +0800]	河南	焦作	电信	2019	09	20
www.baidu.com	139.207.106.29	722052	[09/20/2019 14:33:52 +0800]	四川		电信	2019	09	20
www.alibaba.com	182.91.131.22	22103	[09/20/2019 14:33:52 +0800]	广西	桂林	联通	2019	09	20
www.huawei.com	171.12.12.85	155927	[09/20/2019 14:33:52 +0800]	河南	郑州	电信	2019	09	20

//"""多目录输出:依据不同的key输出不同的数据
//  www.alibaba.com: ip,流量,省,地址
		182.86.190.207  2271 江西  景德镇
//  www.baidu.com：ip,流量,省,时间
		182.86.190.207  2271 江西 [09/20/2019 14:33:52 +0800]
//  www.huawei.com：ip,流量,省,地址,所属营业厅
		182.86.190.207  2271 江西  景德镇 电信	
//  
//"""
代码实现 
val dataRdd = sc.textFile("data/etl_access.log")
      .filter(x=>x.split("t").length==10)
      .map(x=> {
        val splits = x.split("t")
          val domain = splits(0)
          val ip = splits(1)
          val flow = splits(2)
          val pro = splits(4)
          val city = splits(5)
          val isp = splits(6)
          val year = splits(7)
          val month = splits(8)
          val day = splits(9)
        (domain,(domain,ip,flow,pro,city,isp,year,month,day))
      }).coalesce(1,true)
      .saveAsHadoopFile(outpath,classOf[String],classOf[String],classOf[RDDMultipleTextOutputFormat] )
 //注:saveAsHadoopFile 必须是key-value类型 ;saveAsTextFiled底层也是调用saveAsHadoopFile方法的
 
// 重写 MultipleTextOutputFormat 方法:

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String ={
      //已key作为分目录
      val fliename = key+"/"+name
      fliename
    }

    override def generateActualKey(key: Any, value: Any): Any = {
      //输出数据 不需要key
      NullWritable.get()
    }

    override def generateActualValue(key: Any, value: Any): Any = {
      //根据不同的key ,输出不同的value 待优化--->可以从数据库获取想要输出的value
      val splits = value.toString.split(",")
      val sb = new StringBuffer
      key match {
        case "www.ruozedata.com" =>{
          sb.append(splits(0)).append("t")
            .append(splits(1)).append("t")
            .append(splits(3))
          sb
        }
        case "www.baidu.com" =>{
          sb.append(splits(0)).append("t")
            .append(splits(1)).append("t")
            .append(splits(2))
          sb
        }
        case "www.alibaba.com" =>{
          sb.append(splits(0)).append("t")
            .append(splits(2)).append("t")
            .append(splits(3)).append("t")
            .append(splits(4))
          sb
        }
        case _ =>NullWritable.get()
      }
    }

  }