1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package com.gmei
import java.io.Serializable
import com.gmei.WeafareStat.{defaultParams, parser}
import org.apache.spark.sql.{SaveMode, TiContext}
import org.apache.log4j.{Level, Logger}
import scopt.OptionParser
import com.gmei.lib.AbstractParams
import java.io._
object temp_analysis {
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)
case class Params(env: String = "dev",
date: String = "2018-08-01"
) extends AbstractParams[Params] with Serializable
val defaultParams = Params()
val parser = new OptionParser[Params]("Feed_EDA") {
head("WeafareStat")
opt[String]("env")
.text(s"the databases environment you used")
.action((x, c) => c.copy(env = x))
opt[String] ("date")
.text(s"the date you used")
.action((x,c) => c.copy(date = x))
note(
"""
|For example, the following command runs this app on a tidb dataset:
|
| spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
""".stripMargin +
s"| --env ${defaultParams.env}"
)
}
def main(args: Array[String]): Unit = {
parser.parse(args, defaultParams).map { param =>
GmeiConfig.setup(param.env)
val spark_env = GmeiConfig.getSparkSession()
val sc = spark_env._2
val ti = new TiContext(sc)
ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")
import sc.implicits._
val stat_date = GmeiConfig.getMinusNDate(1)
//println(param.date)
val partition_date = stat_date.replace("-","")
//获取策略命中用户device_id
val device_id_cover = sc.sql(
s"""
|select distinct(device_id) as device_id
|from merge_queue_table
""".stripMargin
)
device_id_cover.createOrReplaceTempView("device_id_cover_older")
val diary_pv = sc.sql(
s"""
|select '${stat_date}' as stat_date,count(params["business_id"]) as diary_pv,count(distinct(cl_id)) as device_num_diary
|from online.tl_hdfs_maidian_view
|where action="page_view"
|and params["page_name"]="diary_detail"
|and (params["out"]-params["in"])<7200
|and partition_date ='${partition_date}'
""".stripMargin
)
val meigou_pv = sc.sql(
s"""
|select '${stat_date}' as stat_date,count(params["business_id"]) as meigou_pv,count(distinct(cl_id)) as device_num_meigou
|from online.tl_hdfs_maidian_view
|where action="page_view"
|and params["page_name"]="welfare_detail"
|and (params["out"]-params["in"])<7200
|and partition_date ='${partition_date}'
""".stripMargin
)
val result = diary_pv.join(meigou_pv,"stat_date")
result.show()
GmeiConfig.writeToJDBCTable(result, "diary_pv", SaveMode.Append)
// result.select("stat_date","diary_pv","device_num_diary","meigou_pv","device_num_meigou").write.mode(SaveMode.Append).save("/data2/test.txt")
}
}
}