清掉数据_学习之大数据项目笔记第七篇【数仓模块-日志预处理篇】

1 需求说明

1.1 清洗过滤

1）去除json数据体中的废弃字段（这是前端开发人员在埋点设计方案变更后遗留的无用字段）：

"email"
"phoneNbr"
"birthday"
"isRegistered"
"isLogin"
"addr"
"gender"

3）过滤掉日志中缺少关键字段（event/eventid/sessionid 缺任何一个都不行）的记录！

4）过滤掉json格式不正确的（脏数据）！

1.2 数据解析

将json打平：解析成扁平格式；

注： event字段不用扁平化；转成Map类型存储即可

1.3 数据集成

1）将日志中的GPS经纬度坐标解析成省、市、县（区）信息；（为了方便后续的地域维度分析）

2）集成商圈信息；（为了方便后续的地域维度分析）

1.4 数据修正

1）guid回补

2）字段名称规范化

比如app日志中pgid，wxapp中这个字段叫pageid，和web端日志中的page，统一成pageid

3）字段度量规范化

比如时间戳统一用秒级

4）字段类型规范化

比如时间戳统一用长整型

1.5 保存结果

最后，将数据输出为parquet格式，压缩编码用snappy

2 预处理开发实现

2.1 整体流程

1）json解析，解析成功的返回LogBean对象，解析失败的返回null

（这样一来，json格式不对、不完整的脏数据就被识别出来了）

2）对上一步结果RDD[LogBean]进行过滤（清掉json不完整的脏数据，清掉不符合规则的数据）

3）数据修正（回补uid，统一命名规范、度量单位规范等）

4）对数据进行字典知识集成

5）从集成后的结果中跳出无法解析的gps，写入一个待解析目录

6）输出最终结果保存为parquet（或ORC）文件

3.2 完整代码

case class定义：

case class AppLogBean(var guid:Long,eventid: String,event: Map[String, String],uid: String,imei: String,mac: String,imsi: String,osName: String,osVer: String,androidId: String,resolution: String,deviceType: String,deviceId: String,uuid: String,appid: String,appVer: String,release_ch: String,promotion_ch: String,longtitude: Double,latitude: Double,carrier: String,netType: String,cid_sn: String,ip: String,sessionId: String,timestamp: Long,var province:String="未知",var city:String="未知",var district:String="未知")

预处理流程如下：

package cn.doitedu.dw.preimport java.utilimport ch.hsr.geohash.GeoHash
import cn.doitedu.commons.util.SparkUtil
import cn.doitedu.dw.beans.AppLogBean
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{Dataset, Row}object AppLogDataPreprocess {def main(args: Array[String]): Unit = {// 构造sparksessiongval spark = SparkUtil.getSparkSession(this.getClass.getSimpleName)import spark.implicits._// 加载当日的app埋点日志文件，成为一个dataset[String]val appDs: Dataset[String] = spark.read.textFile("G:yiee_logs2020-01-12app")// 加载geo地域字典数据/***    -----|---------|------|----------|*    geo  |province |  city|  district|*    -----|---------|------|----------|*    39eu |河北省    | 石家庄| 裕华区    | Row*    y67u |河南省    | 郑州市| 金水区    | Row*/val geodf = spark.read.parquet("data/dict/geo_dict/output")val geoMap: collection.Map[String, (String, String, String)] = geodf.rdd.map(row=>{val geo = row.getAs[String]("geo")val province = row.getAs[String]("province")val city = row.getAs[String]("city")val district = row.getAs[String]("district")(geo,(province,city,district))}).collectAsMap()// 广播地域字典// Map{ 39eu -> (河北省,石家庄,裕华区)//      y67u -> (河南省,郑州市,金水区)//    }val bc_geo = spark.sparkContext.broadcast(geoMap)// 加载id映射字典/***    ---------------|------|* biaoshi_hashcode  |  guid|*    ---------------|------|*     8238574359     | 62375|row*    ---------------|------|*     3285943259     | 62375|row*    ---------------|------|*          62375      | 62375|row*  -----------------|------|*/val idmpdf = spark.read.parquet("data/idmp/2020-01-12")val idMap = idmpdf.rdd.map(row=>{val id = row.getAs[Long]("biaoshi_hashcode")val guid = row.getAs[Long]("guid")(id,guid)}).collectAsMap()val bc_id = spark.sparkContext.broadcast(idMap)// 对日志ds集合中的每一条记录（json）进行解析appDs.map(line => {var bean: AppLogBean = nulltry {val jsonobj = JSON.parseObject(line)val eventid = jsonobj.getString("eventid")val timestamp = jsonobj.getString("timestamp").toLongval eventobj: JSONObject = jsonobj.getJSONObject("event")import scala.collection.JavaConversions._val javaMap: util.Map[String, String] = eventobj.getInnerMap.asInstanceOf[util.Map[String, String]]val event: Map[String, String] = javaMap.toMapval userobj = jsonobj.getJSONObject("user")val uid = userobj.getString("uid")val sessionId = userobj.getString("sessionId")val phoneobj = userobj.getJSONObject("phone")val imei = phoneobj.getString("imei")val mac = phoneobj.getString("mac")val imsi = phoneobj.getString("imsi")val osName = phoneobj.getString("osName")val osVer = phoneobj.getString("osVer")val androidId = phoneobj.getString("androidId")val resolution = phoneobj.getString("resolution")val deviceType = phoneobj.getString("deviceType")val deviceId = phoneobj.getString("deviceId")val uuid = phoneobj.getString("uuid")val appobj = jsonobj.getJSONObject("app")val appid = appobj.getString("appid")val appVer = appobj.getString("appVer")val release_ch = appobj.getString("release_ch") // 下载渠道val promotion_ch = appobj.getString("promotion_ch") // 推广渠道val locobj = jsonobj.getJSONObject("loc")var lng = 0.0var lat = -90.0try {lng = locobj.getDouble("longtitude")lat = locobj.getDouble("latitude")} catch {case e: Exception =>}val carrier = locobj.getString("carrier")val netType = locobj.getString("netType")val cid_sn = locobj.getString("cid_sn")val ip = locobj.getString("ip")// 判断数据合法规则val tmp = (imei + imsi + mac + uid + uuid + androidId).replaceAll("null", "")if (StringUtils.isNotBlank(tmp) && event != null && StringUtils.isNotBlank(eventid) && StringUtils.isNotBlank(sessionId)) {// 将提取出来的各个字段，封装到AppLogBean中bean = AppLogBean(Long.MinValue,eventid,event,uid,imei,mac,imsi,osName,osVer,androidId,resolution,deviceType,deviceId,uuid,appid,appVer,release_ch,promotion_ch,lng,lat,carrier,netType,cid_sn,ip,sessionId,timestamp)}} catch {case e: Exception => null}bean}).filter(_ != null).map(bean=>{val geoDict = bc_geo.valueval idmpDict = bc_id.value// 查geo地域字典，填充省市区val lat = bean.latitudeval lng = bean.longtitudeval mygeo = GeoHash.geoHashStringWithCharacterPrecision(lat,lng,5)val maybeTuple: Option[(String, String, String)] = geoDict.get(mygeo)if(maybeTuple.isDefined){val areaNames = maybeTuple.get// 填充省市区bean.province = areaNames._1bean.city = areaNames._2bean.district = areaNames._3}// 查id映射字典，填充guidval ids = Array(bean.imei,bean.imsi,bean.mac,bean.androidId,bean.uuid,bean.uid)val mouId = ids.filter(StringUtils.isNotBlank(_))(0)val maybeLong = idmpDict.get(mouId.hashCode.toLong)if(maybeLong.isDefined){val guid = maybeLong.getbean.guid = guid}bean}).filter(bean=>bean.guid != Long.MinValue).toDF().write.parquet("data/applog_processed/2020-01-12")spark.close()}}

2.3 打包提交线上运行

步骤1：

将代码中写死的路径换成参数形式

步骤2：

在pom中加入打包插件

<build><plugins><!-- 指定编译java的插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.5.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><!-- 指定编译scala的插件 --><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals><configuration><args><arg>-dependencyfile</arg><arg>${project.build.directory}/.scala_dependencies</arg></args></configuration></execution></executions></plugin></plugins>
</build>

步骤3：

在idea的maven侧边栏卡中，选父工程，点击install，对整个工程进行打包和本地库安装

步骤4：

拷贝预处理程序的jar包上传到集群，用命令提交

 bin/spark-submit  --master yarn --deploy-mode cluster --num-executors 3 --executor-memory 1g --executor-cores 1 --class cn.doitedu.titan.dw.pre.AppEventLogPreprocess /root/dw.jar /titan/applog/2019-10-29 /titan/areadict /titan/output/applog/2019-10-29 yarn