package mobvista.dmp.datasource.ga
import java.security.{MessageDigest, NoSuchAlgorithmException}
import java.text.SimpleDateFormat
import java.util.regex.Pattern
import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.types.{StringType, StructField, StructType}
/**
* @package: mobvista.dmp.datasource
* @author: wangjf
* @date: 2018/11/3
* @time: 下午2:44
* @email: jinfeng.wang@mobvista.com
* @phone: 152-1062-7698
*/
object Constant {
def schema: StructType = {
StructType(StructField("text", StringType) ::
Nil)
}
val event_sql: String =
s"""
|INSERT OVERWRITE TABLE dev.event_@date_str_undline
| SELECT json_col(`text`,'data','v') AS version,
| json_col(`text`,'game_id','') AS game_id,
| json_col(`text`,'data','user_id') AS user_id,
| json_col(`text`,'arrival_ts','') AS arrival_ts,
| json_col(`text`,'user_meta','install_ts') AS install_ts,
| COALESCE(json_col(`text`,'category',''),json_col(`text`,'data','category')) AS category,
| json_col(`text`,'data','session_id') AS session_id,
| json_col(`text`,'data','event_id') AS event_id,
| json_col(`text`,'country_code','') AS country_code,
| json_col(`text`,'ip','') AS ip,
| json_col(`text`,'user_meta','revenue') AS revenue,
| json_col(`text`,'data','facebook_id') AS facebook_id,
| json_col(`text`,'data','googleplus_id') AS googleplus_id,
| json_col(`text`,'data','ios_id') AS ios_id,
| json_col(`text`,'data','android_id') AS android_id,
| json_col(`text`,'data','google_aid') AS google_aid,
| json_col(`text`,'data','ios_idfa') AS ios_idfa,
| json_col(`text`,'data','ios_idfv') AS ios_idfv,
| json_col(`text`,'data','limit_ad_tracking') AS is_ad_tracking_limited,
| json_col(`text`,'data','logon_gamecenter') AS is_logged_gamecenter,
| json_col(`text`,'data','logon_googleplay') AS is_logged_googleplay,
| json_col(`text`,'data','value') AS value,
| json_col(`text`,'data','attempt_num') AS attempt_num,
| case when json_col(`text`,'data','score') >= 2147483647 then 0 else json_col(`text`,'data','score') end as score,
| json_col(`text`,'data','session_num') AS session_num,
| json_col(`text`,'data','length') AS session_length,
| json_col(`text`,'data','gender') AS gender,
| json_col(`text`,'data','birth_year') AS birth_year,
| json_col(`text`,'data','progression') AS progression,
| json_col(`text`,'data','sdk_version') AS sdk_version,
| json_col(`text`,'data','engine_version') AS engine_version,
| json_col(`text`,'data','os_version') AS os_version,
| json_col(`text`,'data','manufacturer') AS manufacturer,
| json_col(`text`,'data','device') AS device,
| json_col(`text`,'data','platform') AS platform,
| json_col(`text`,'data','build') AS build,
| json_col(`text`,'data','connection_type') AS connection_type,
| json_col(`text`,'data','jailbroken') AS is_jailbroken,
| json_col(`text`,'user_meta','pay_ft') AS pay_ft,
| json_col(`text`,'data','currency') AS currency,
| json_col(`text`,'data','amount') AS amount,
| json_col(`text`,'data','cart_type') AS cart_type,
| json_col(`text`,'data','transaction_num') AS transaction_num,
| json_col(`text`,'user_meta','receipt_status') AS receipt_status,
| json_col(`text`,'user_meta','os_major') AS v1_os_major,
| json_col(`text`,'user_meta','device') AS v1_device,
| json_col(`text`,'user_meta','platform') AS v1_platform,
| json_col(`text`,'user_meta','gender') AS v1_gender,
| json_col(`text`,'user_meta','sdk_version') AS v1_sdk_version,
| json_col(`text`,'user_meta','google_aid') as v1_google_aid,
| json_col(`text`,'user_meta','android_id') as v1_android_id,
| json_col(`text`,'data','category') AS v2_category,
| json_col(`text`,'user_meta','install_publisher') AS install_publisher,
| json_col(`text`,'user_meta','install_site') AS install_site,
| json_col(`text`,'user_meta','install_campaign') AS install_campaign,
| json_col(`text`,'user_meta','install_adgroup') AS install_adgroup,
| json_col(`text`,'user_meta','install_ad') AS install_ad,
| json_col(`text`,'user_meta','install_keyword') AS install_keyword,
| json_col(`text`,'data','severity') AS severity
|FROM event_text
|WHERE to_date(json_col(`text`,'arrival_ts','')) LIKE '@date_str_midline' AND json_col(`text`,'data','user_id') != '00000000-0000-0000-0000-000000000000' AND json_col(`text`,'data','user_id') != '0'
| AND json_col(`text`,'data','user_id') IS NOT NULL AND json_col(`text`,'data','user_id') != 'nullUser'
| AND json_col(`text`,'data','user_id') != ''
""".stripMargin
def parserGa(iters: Iterator[String], date: String, filterGameIds: Set[Int]): Iterator[GA] = {
iters.map(iter => {
try {
val mapper = new ObjectMapper()
val jsonNode = mapper.readTree(iter)
val dataNode = jsonNode.path("data")
val countryCodeNodeTmp = jsonNode.path("country_code")
var countryCodeNode = countryCodeNodeTmp.asText("")
if (StringUtils.isNotBlank(countryCodeNodeTmp.asText()) && countryCodeNodeTmp.asText().equalsIgnoreCase("GB")) { //当为GB时要修改成UK
countryCodeNode = "UK"
}
val arrivalTsNode = jsonNode.path("arrival_ts")
val gameIdNode = jsonNode.path("game_id")
val ipNode = jsonNode.path("ip")
val categoryNode = jsonNode.path("category")
val userMetaNode = jsonNode.path("user_meta")
val category = if (StringUtils.isNotBlank(categoryNode.asText())) {
categoryNode.asText()
} else {
dataNode.path(Data.CATEGORY.toString).asText()
}
val score: Integer = if (!dataNode.has(Data.SCORE.toString)) {
null
} else if (dataNode.path(Data.SCORE.toString).asInt() > 2147483647) {
0
} else {
dataNode.path(Data.SCORE.toString).asInt()
}
val game_id = if (StringUtils.isNotBlank(dataNode.path(Data.GAME_ID.toString).asText())) {
dataNode.path(Data.GAME_ID.toString).asInt()
} else {
gameIdNode.asInt()
}
val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
val arrival_ts = format.format(new java.util.Date(arrivalTsNode.asLong(0) * 1000))
val install_ts = format.format(userMetaNode.path(UserMeta.INSTALL_TS.toString).asLong(0) * 1000)
val revenue = userMetaNode.path(UserMeta.REVENUE.toString).toString
val user_id = dataNode.path(Data.USER_ID.toString).asText()
if (arrival_ts.startsWith(date) && check_userId(user_id) && !filterGameIds.contains(game_id)) {
val user_id = generateJsonVal(dataNode, Data.USER_ID.toString)
GA(dataNode.path(Data.VERSION.toString).asInt(), game_id, user_id, md5Str(game_id + user_id), arrival_ts, install_ts, category.trim,
generateJsonVal(dataNode, Data.SESSION_ID.toString), generateJsonVal(dataNode, Data.EVENT_ID.toString), countryCodeNode.trim, ipNode.asText().trim,
revenue.trim, generateJsonVal(dataNode, Data.FACEBOOK_ID.toString), generateJsonVal(dataNode, Data.GOOGLEPLUS_ID.toString), generateJsonVal(dataNode, Data.IOS_ID.toString),
generateJsonVal(dataNode, Data.ANDROID_ID.toString), generateJsonVal(dataNode, Data.GOOGLE_AID.toString), generateJsonVal(dataNode, Data.IOS_IDFA.toString),
generateJsonVal(dataNode, Data.IOS_IDFV.toString), dataNode.path(Data.LIMIT_AD_TRACKING.toString).asBoolean(), dataNode.path(Data.LOGON_GAMECENTER.toString).asBoolean(),
dataNode.path(Data.LOGON_GOOGLEPLAY.toString).asBoolean(), dataNode.path(Data.VALUE.toString).asDouble(), dataNode.path(Data.ATTEMPT_NUM.toString).asInt(),
score, dataNode.path(Data.SESSION_NUM.toString).asInt(), dataNode.path(Data.LENGTH.toString).asInt(), generateJsonVal(dataNode, Data.GENDER.toString),
generateJsonVal(dataNode, Data.BIRTH_YEAR.toString), generateJsonVal(dataNode, Data.PROGRESSION.toString), generateJsonVal(dataNode, Data.SDK_VERSION.toString),
generateJsonVal(dataNode, Data.ENGINE_VERSION.toString), generateJsonVal(dataNode, Data.OS_VERSION.toString), generateJsonVal(dataNode, Data.MANUFACTURER.toString),
generateJsonVal(dataNode, Data.DEVICE.toString), generateJsonVal(dataNode, Data.PLATFORM.toString), generateJsonVal(dataNode, Data.BUILD.toString),
generateJsonVal(dataNode, Data.CONNECTION_TYPE.toString), dataNode.path(Data.JAILBROKEN.toString).asBoolean(), generateJsonVal(userMetaNode, UserMeta.PAY_FT.toString),
generateJsonVal(dataNode, Data.CURRENCY.toString), dataNode.path(Data.AMOUNT.toString).asDouble(), generateJsonVal(dataNode, Data.CART_TYPE.toString),
dataNode.path(Data.TRANSACTION_NUM.toString).asInt(), generateJsonVal(userMetaNode, UserMeta.RECEIPT_STATUS.toString), generateJsonVal(userMetaNode, UserMeta.OS_MAJOR.toString),
generateJsonVal(userMetaNode, UserMeta.DEVICE.toString), generateJsonVal(userMetaNode, UserMeta.PLATFORM.toString), generateJsonVal(userMetaNode, UserMeta.GENDER.toString),
generateJsonVal(userMetaNode, UserMeta.SDK_VERSION.toString), generateJsonVal(userMetaNode, UserMeta.GOOGLE_AID.toString), generateJsonVal(userMetaNode, UserMeta.ANDROID_ID.toString),
generateJsonVal(dataNode, Data.CATEGORY.toString), generateJsonVal(userMetaNode, UserMeta.INSTALL_PUBLISHER.toString), generateJsonVal(userMetaNode, UserMeta.INSTALL_SITE.toString),
generateJsonVal(userMetaNode, UserMeta.INSTALL_CAMPAIGN.toString), generateJsonVal(userMetaNode, UserMeta.INSTALL_ADGROUP.toString), generateJsonVal(userMetaNode, UserMeta.INSTALL_AD.toString),
generateJsonVal(userMetaNode, UserMeta.INSTALL_KEYWORD.toString), generateJsonVal(dataNode, Data.SEVERITY.toString)
)
} else {
null
}
} catch {
case _: Exception =>
null
}
})
}
val allZero = "00000000-0000-0000-0000-000000000000"
private val pattern = Pattern.compile("^[a-zA-Z0-9_-]+$")
def generateJsonVal(dataNode: com.fasterxml.jackson.databind.JsonNode, key: String): String = {
if (StringUtils.isNotBlank(dataNode.path(key).asText())) {
dataNode.path(key).asText().trim.replace("|", "").replace("\t", "")
} else {
null
}
}
/* def md5Str(s: String) = {
val m = java.security.MessageDigest.getInstance("MD5")
val b = s.getBytes("UTF-8")
m.update(b,0,b.length)
new java.math.BigInteger(1,m.digest()).toString(16)
}*/
def md5Str(str: String) = {
val hexString = new StringBuffer
try {
val md = MessageDigest.getInstance("MD5")
md.update(str.getBytes())
val hash = md.digest()
for (i <- 0 until hash.length) {
if ((0xff & hash(i)) < 0x10) {
hexString.append("0" + Integer.toHexString((0xFF & hash(i))));
} else {
hexString.append(Integer.toHexString(0xFF & hash(i)));
}
}
} catch {
case e: NoSuchAlgorithmException => {
e.printStackTrace()
}
}
hexString.toString();
}
def check_userId(user_id: String): Boolean = {
StringUtils.isNotBlank(user_id) && !user_id.equals(allZero) && !user_id.equals("0") && !user_id.equals("nullUser") &&
!user_id.equals("null") && pattern.matcher(user_id).matches()
}
object Data extends Enumeration {
type Data = Value
val VERSION: Data.Value = Value("v")
val GAME_ID: Data.Value = Value("game_id")
val USER_ID: Data.Value = Value("user_id")
val GOOGLE_AID: Data.Value = Value("google_aid")
val SDK_VERSION: Data.Value = Value("sdk_version")
val OS_VERSION: Data.Value = Value("os_version")
val MANUFACTURER: Data.Value = Value("manufacturer")
val DEVICE: Data.Value = Value("device")
val PLATFORM: Data.Value = Value("platform")
val SESSION_ID: Data.Value = Value("session_id")
val SESSION_NUM: Data.Value = Value("session_num")
val CONNECTION_TYPE: Data.Value = Value("connection_type")
val ENGINE_VERSION: Data.Value = Value("engine_version")
val BUILD: Data.Value = Value("build")
val EVENT_ID: Data.Value = Value("event_id")
val CATEGORY: Data.Value = Value("category")
val CURRENCY: Data.Value = Value("currency")
val AMOUNT: Data.Value = Value("amount")
val TRANSACTION_NUM: Data.Value = Value("transaction_num")
val CART_TYPE: Data.Value = Value("cart_type")
val IOS_IDFV: Data.Value = Value("ios_idfv")
val IOS_IDFA: Data.Value = Value("ios_idfa")
val VALUE: Data.Value = Value("value")
val SCORE: Data.Value = Value("score")
val ATTEMPT_NUM: Data.Value = Value("attempt_num")
val LENGTH: Data.Value = Value("length")
val FACEBOOK_ID: Data.Value = Value("facebook_id")
val GOOGLEPLUS_ID: Data.Value = Value("googleplus_id")
val IOS_ID: Data.Value = Value("ios_id")
val ANDROID_ID: Data.Value = Value("android_id")
val LIMIT_AD_TRACKING: Data.Value = Value("limit_ad_tracking")
val LOGON_GAMECENTER: Data.Value = Value("logon_gamecenter")
val LOGON_GOOGLEPLAY: Data.Value = Value("logon_googleplay")
val GENDER: Data.Value = Value("gender")
val BIRTH_YEAR: Data.Value = Value("birth_year")
val PROGRESSION: Data.Value = Value("progression")
val JAILBROKEN: Data.Value = Value("jailbroken")
val SEVERITY: Data.Value = Value("severity")
}
object UserMeta extends Enumeration {
type UserMeta = Value
val INSTALL_TS: UserMeta.Value = Value("install_ts")
val REVENUE: UserMeta.Value = Value("revenue")
val RECEIPT_STATUS: UserMeta.Value = Value("receipt_status")
val PAY_FT: UserMeta.Value = Value("pay_ft")
val SDK_VERSION: UserMeta.Value = Value("sdk_version")
val OS_MAJOR: UserMeta.Value = Value("os_major")
val DEVICE: UserMeta.Value = Value("device")
val PLATFORM: UserMeta.Value = Value("platform")
val GENDER: UserMeta.Value = Value("gender")
val GOOGLE_AID: UserMeta.Value = Value("google_aid")
val ANDROID_ID: UserMeta.Value = Value("android_id")
val INSTALL_PUBLISHER: UserMeta.Value = Value("install_publisher")
val INSTALL_SITE: UserMeta.Value = Value("install_site")
val INSTALL_CAMPAIGN: UserMeta.Value = Value("install_campaign")
val INSTALL_ADGROUP: UserMeta.Value = Value("install_adgroup")
val INSTALL_AD: UserMeta.Value = Value("install_ad")
val INSTALL_KEYWORD: UserMeta.Value = Value("install_keyword")
}
// ods_ga_active_total
val ods_ga_active_total_sql: String =
"""
|SELECT device_id, platform
| FROM dwh.ods_ga_active_total WHERE dt = '@ga_date'
""".stripMargin
val dm_interest_tag_sql: String =
"""
|SELECT device_id, platform
| FROM dwh.dm_interest_tag_orc WHERE CONCAT(year,month,day) = '@date' AND business = 'ga'
""".stripMargin
}