1. log4j（具体见log4j文档）
log4j是一个java系统中用于输出日志信息的工具。log4j可以将日志定义成多种级别：ERROR / WARN / INFO / DEBUG
log4j通过获取到一个logger对象来输出日志：
val logger = Logger.getLogger("logger名称"); logger.info("日志内容")
所拿到的这些logger对象之间是有“父子”关系的，所有logger都是rootLogger的子！
"org.apache" 这个名字的logger是 "org"这个名字的logger的子！
log4j的日志输出格式和目的地，都是可以通过参数配置的；
-
目的地的控制用Appender输出组件
常用的Appender组件：
log4j.appender.xx=org.apache.log4j.ConsoleAppender
log4j.appender.rollingFile=org.apache.log4j.RollingFileAppender
-
格式的控制用LayOut布局组件
log4j.appender.xx.layout=org.apache.log4j.PatternLayout
log4j.appender.xx.layout.ConversionPattern=[%-5p] %d(%r) --> [%t] %l: %m %x %n
2. 父子maven工程
（1）创建一个父工程（如平常创建一样），父工程中不写代码，所以最好将src文件夹删除（比如公司新手会将代码误写入该文件夹）
（2）创建子工程
得到如下图
接着如下所示
到此，一个子maven项目dataware即建立成功，子项目的pom文件如下所示
若是子工程中的父工程配置删除后，子工程不认识父工程，但是父工程认识子工程
（3）说明
A. 父工程pom文件中引入公共的依赖和插件（会被子工程pom继承）,此处有几处规范
- 依赖定义的管理（不是真正引入依赖） 标签:<dependencyManagement><dependencyManagement>
作用：父项目中某个子项目需要用到某个依赖，这个时候若是在子项目的pom文件中定义这个依赖的版本，当另外一个子项目也要这个依赖时，由于需要统一依赖的版本，这时另外一个子项目中也需要定义相同版本的依赖。这样就比较麻烦，这个时候就可以使用依赖定义的管理（在父工程中定义子项目需要依赖的版本，子项目中就不需要写依赖的版本），如下
父工程pom文件（部分）
<dependencyManagement>
<dependency>
<groupId>ch.hsr</groupId>
<artifactId>geohash</artifactId>
<version>1.3.0</version>
</dependency>
</dependencies>
</dependencyManagement>
子工程pom文件
- 属性定义（标签：<properties><properties>）
- 依赖排除(标签：<exclusions><exclusions>): 解决jar包的版本冲突
比如下面的spark使用的hadoop版本就出现依赖的冲突
解决办法（排除依赖）
<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark.version}</version> <exclusions> <exclusion> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> </exclusion> </exclusions> </dependency>
- 当在idea删除某个项目时，再创建一个同名的项目时，会出错（Idea中记录的东西会冲突）
解决办法：
直接到项目的目录中将idea的相关文件删除掉，如下图所示
spring子项目的创建
3.项目开发（埋点日志预处理-json数据解析、清洗过滤、数据集成实现、uid回补）
3.1 json数据格式如下：
3.2 需求说明
3.2.1 清洗过滤
此处为了记录数据方便，定义一个AppLogBean，该类中定义了两个方法（1.解析json 返回一个case class， 2. 判断一个bean是否有效），并在该类中定义一个case class AppLogBean
AppLogBean代码
package com._51doit.tian.dw.pre import com.alibaba.fastjson.{JSON, JSONObject} import org.apache.commons.lang3.StringUtils import scala.collection.mutable case class AppLogBean( eventid :String , timestamp :Double , event :Map[String,String] , uid :String , phoneNbr :String , sessionId :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, var province:String = "", var city:String = "", var district:String = "", var dateStr:String = "", var timeStr:String = "" ) object AppLogBean { /** * 解析app埋点json日志，返回一个case class */ def parseJson2Bean(line:String): AppLogBean ={ try { val obj: JSONObject = JSON.parseObject(line) val eventid: String = obj.getString("eventid") val timestamp = obj.getString("timestamp").toDouble val event: JSONObject = obj.getJSONObject("event") val eventMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]() import scala.collection.JavaConversions._ for(ent <- event.entrySet()){ eventMap.put(ent.getKey,ent.getValue.toString) } val user = obj.getJSONObject("user") val uid = user.getString("uid") val phoneNbr = user.getString("phoneNbr") val sessionId = user.getString("sessionId") val phone = user.getJSONObject("phone") val imei = phone.getString("imei") val mac = phone.getString("mac") val imsi = phone.getString("imsi") val osName = phone.getString("osName") val osVer = phone.getString("osVer") val androidId = phone.getString("androidId") val resolution = phone.getString("resolution") val deviceType = phone.getString("deviceType") val deviceId = phone.getString("deviceId") val uuid = phone.getString("uuid") val app = user.getJSONObject("app") val appid = app.getString("appid") val appVer = app.getString("appVer") val release_ch = app.getString("release_ch") val promotion_ch = app.getString("promotion_ch") val loc = user.getJSONObject("loc") val longtitude = loc.getDouble("longtitude") val latitude = loc.getDouble("latitude") val carrier = loc.getString("carrier") val netType = loc.getString("netType") val cid_sn = loc.getString("cid_sn") val ip = loc.getString("ip") AppLogBean( eventid , timestamp, eventMap.toMap, uid , phoneNbr , sessionId , imei , mac , imsi , osName , osVer , androidId , resolution , deviceType , deviceId , uuid , appid , appVer , release_ch , promotion_ch , longtitude , latitude , carrier , netType , cid_sn , ip ) } catch { case e: Exception => null case _: Throwable => null } } /** * 判断一条bean是否有效 */ def isValidBean(bean:AppLogBean): Boolean ={ val uid: String = bean.uid val imei: String = bean.imei val uuid: String = bean.uuid val mac: String = bean.mac val androidId: String = bean.androidId val ip: String = bean.ip // 以上参数不能全为空 var flag1 = StringUtils.isNotBlank((uid + imei + uuid + mac + androidId + ip).replaceAll("null", "")) val event: Map[String, String] = bean.event val eventid: String = bean.eventid val sessionId = bean.sessionId var flag2 = (event != null) && (StringUtils.isNotBlank(eventid) ) && (StringUtils.isNotBlank(sessionId)) flag1 && flag2 } }
3.2.2 数据解析
此处event数据不用扁平化的原因是，event内的数据类型也不一样
3.2.3 数据集成
3.2.4 数据修正
思路图
注意：此处手机标识 比如imei为空时，作为join on相等的条件时会出错，一定要判断非空，由于sql语句很麻烦（如下），所以开发一个自定义函数，用来判断两个字符串在非空情况下是否相等
每一个手机识别方式都要这样写，很麻烦，以下是自定义的函数
// 开发一个自定义函数，用来判断两个字符串在非空情况下是否相等 val is_equal = (x: String, y: String) => { if (x != y || StringUtils.isBlank(x) || StringUtils.isBlank(y)) false else true } spark.udf.register("is_equal",is_equal)
业务代码
AppEventLogPreprocess
package com._51doit.tian.dw.pre import java.text.SimpleDateFormat import ch.hsr.geohash.GeoHash import com._51doit.tian.commons.utils.{DictLoadUtil, SparkUtils} import org.apache.commons.lang3.StringUtils import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} object AppEventLogPreprocess { def main(args: Array[String]): Unit = { Logger.getLogger("org").setLevel(Level.WARN) val spark: SparkSession = SparkUtils.getSpark(this.getClass.getSimpleName) import spark.implicits._ // 加载原始日志文件 val ds: Dataset[String] = spark.read.textFile("E:\javafile\dataware\2019-10-29") // 解析json val beans: Dataset[AppLogBean] = ds.map(AppLogBean.parseJson2Bean) /** * 清洗过滤 */ val validBeans: Dataset[AppLogBean] = beans .filter(_ != null) .filter(AppLogBean.isValidBean(_)) /** * 数据集成 */ val dictDF: DataFrame = spark.read.parquet("E:/javafile/spark/out11") val dictMap: collection.Map[String, (String, String, String)] = DictLoadUtil.loadAreaDict(dictDF) val bc_dict = spark.sparkContext.broadcast(dictMap) // 然后进行集成 val integrated: Dataset[AppLogBean] = validBeans.mapPartitions(iter => { // 取广播变量 val dict: collection.Map[String, (String, String, String)] = bc_dict.value iter.map(bean => { // 处理GPS坐标 val longtitude: Double = bean.longtitude val latitude: Double = bean.latitude // 如果经纬度坐标在中国的经纬度范围之内，才去转geohash编码并从字典中查找省市区 if (longtitude > 0 && longtitude < 120 && latitude > 0 && latitude < 70) { val geo = GeoHash.withCharacterPrecision(latitude, longtitude, 5).toBase32 val area = dict.getOrElse(geo, ("", "", "")) bean.province = area._1 bean.city = area._2 bean.district = area._3 } // 处理时间戳 val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") val str: Array[String] = sdf.format(bean.timestamp).split(" ") bean.dateStr = str(0) bean.timeStr = str(1) // 返回集成完成的bean bean }) }) // integrated.where("trim(province) != '' ").show(10,false) /** * 数据修正 */ val haveUid = integrated.where("uid is not null and trim(uid) !='' ") val noUid = integrated.where(" uid is null or trim(uid) ='' ") import org.apache.spark.sql.functions._ val uids = haveUid .groupBy($"uid") .agg( max("imei").as("imei"), max("imsi").as("imsi"), max("mac").as("mac"), max("uuid").as("uuid"), max("androidId").as("androidId"), max("deviceId").as("deviceId") ) noUid.createTempView("nouid") uids.createTempView("uids") // 开发一个自定义函数，用来判断两个字符串在非空情况下是否相等 val is_equal = (x: String, y: String) => { if (x != y || StringUtils.isBlank(x) || StringUtils.isBlank(y)) false else true } spark.udf.register("is_equal",is_equal) // 对没有uid的数据进行回补操作 val part1: DataFrame = spark.sql( """ | |select | |a.eventid , |a.timestamp , |a.event , |if(b.uid is not null,b.uid,a.uid) as uid, |a.phoneNbr , |a.sessionId , |a.imei , |a.mac , |a.imsi , |a.osName , |a.osVer , |a.androidId , |a.resolution , |a.deviceType , |a.deviceId , |a.uuid , |a.appid , |a.appVer , |a.release_ch , |a.promotion_ch , |a.longtitude , |a.latitude , |a.carrier , |a.netType , |a.cid_sn , |a.ip , |a.province, |a.city, |a.district, |a.dateStr, |a.timeStr | |from | |nouid a left join uids b | on is_equal(a.imei,b.imei) | or is_equal(a.imsi,b.imsi) | or is_equal(a.mac,b.mac) | or is_equal(a.uuid,b.uuid) | or is_equal(a.androidId,b.androidId) | or is_equal(a.deviceId,b.deviceId) | """.stripMargin) // 将回补好的数据 union 原来就有uid的数据 val result = haveUid.toDF.union(part1) // 输出结果 result.write.parquet("E:\javafile\dataware1\2019-10-29") spark.close() } }