Spark- 数据清洗

 

输入输出转化工具类

复制代码
package com.rz.mobile_tag.log



import org.apache..sql.Row
import org.apache..sql.types.{LongType, StringType, StructField, StructType}

/**
  * 访问日志转换(输入==>输出)工具类
  */
object AccessConvertUtil {
  // 定义的输出字段
  val structType = StructType(
    Array(
      StructField("url", StringType),
      StructField("cmsType", StringType),
      StructField("cmsId", LongType),
      StructField("traffic", LongType),
      StructField("ip", StringType),
      StructField("city", StringType),
      StructField("time", StringType),
      StructField("day", StringType)

    )
  )


  /**
    * 根据输入的每一行信息转换成输出的样式
    * @param log 输入的每一行记录信息
    */
  def parseLog(log:String)={
    try{
      val splits = log.split("\t",-1)

      val url = splits(1)
      val traffic = splits(2).toLong
      val ip = splits(3)

      val domain = "http://www.rz.com/"
      val cms = url.substring(url.indexOf(domain)+domain.length)
      val cmsTypeId = cms.split("/")

      var cmsType = ""
      var cmsId = 0l
      if (cmsTypeId.length>1){
        cmsType  = cmsTypeId(0)
        cmsId = cmsTypeId(1).toLong
      }

      val city=""
      val time = splits(0)
      val day = time.substring(0, 10).replaceAll("-","")

      // 这个Row里面的字段要和Struct中的字段对应上
      Row(url, cmsType, cmsId, traffic, ip, city, time, day)
    }catch {
      case e:Exception =>{
        Row(0)
      }
    }
  }


}
复制代码

读取数据,清洗输出目标数据

复制代码
package com.rz.mobile_tag.log

import org.apache..sql.{SaveMode, SparkSession}

/**
  * 使用Spark完成我们的数据清洗操作
  */
object SparkStatCleanJob {
  def main(args: Array[String]): Unit = {
    val  = SparkSession.builder().appName(s"${this.getClass.getSimpleName}")
      .master("local[2]")
      .getOrCreate()

    val accessRDD = .sparkContext.textFile(args(0))
    // debug查看数据
    // accessRDD.take(10).foreach(println)


    val accessDF = .createDataFrame(accessRDD.map(log =>AccessConvertUtil.parseLog(log)),AccessConvertUtil.structType)

    // accessDF.printSchema()
    // accessDF.show(false)
    accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save(args(1))




    .stop()
  }
}
复制代码

 

 

 

人已赞赏
博客

Spark- 求最受欢迎的TopN课程

2019-8-18 18:38:12

博客

Hadoop- HDFS的API操作

2019-8-18 18:41:05

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
有新消息 消息中心
搜索