Commit c593113d authored by 王志伟's avatar 王志伟

重新计算彦钊昨天数据

parent 381617c4
...@@ -51,8 +51,8 @@ object temp_count { ...@@ -51,8 +51,8 @@ object temp_count {
ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table") ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")
// val stat_date = GmeiConfig.getMinusNDate(1) val stat_date = GmeiConfig.getMinusNDate(1)
val stat_date = param.date // val stat_date = param.date
//println(param.date) //println(param.date)
val partition_date = stat_date.replace("-","") val partition_date = stat_date.replace("-","")
val decive_id_oldUser = sc.sql( val decive_id_oldUser = sc.sql(
...@@ -241,6 +241,93 @@ object Repeated_content_recommendation { ...@@ -241,6 +241,93 @@ object Repeated_content_recommendation {
GmeiConfig.writeToJDBCTable(df, table = "Repeated_evaluation_indicator", SaveMode.Append) GmeiConfig.writeToJDBCTable(df, table = "Repeated_evaluation_indicator", SaveMode.Append)
// val temp=get_result.collect()
// for (i <- 0 until 30 ) {
// println(temp(i))
// }
}
}
}
object Repeated_content_recommendation_moreday {
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 = "data_feed_exposure_precise")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")
val stat_date = GmeiConfig.getMinusNDate(1)
// val stat_date = param.date
val partition_date = stat_date.replace("-","")
val exp_diary = sc.sql(
s"""
|select concat_ws('|',device_id,cid_id)
|from data_feed_exposure
|where cid_type = 'diary'
|and device_id not in (select device_id from blacklist)
|and stat_date ='${stat_date}'
""".stripMargin
)
exp_diary.show()
val get_result =exp_diary.rdd.map((_, 1)).reduceByKey(_ + _)
.sortBy(_._2,false)
val more_than2=get_result.filter(_._2 >=2).map(_._2).reduce((x,y)=>x+y)
println(more_than2)
val all =get_result.map(_._2).reduce((x,y)=>x+y)
println(all)
val repeated_rate= more_than2 / all.toDouble
println(repeated_rate)
val test=List((stat_date,repeated_rate))
val df = sc.createDataFrame(test)
GmeiConfig.writeToJDBCTable(df, table = "Repeated_evaluation_indicator_moreday", SaveMode.Append)
// val temp=get_result.collect() // val temp=get_result.collect()
// for (i <- 0 until 30 ) { // for (i <- 0 until 30 ) {
// println(temp(i)) // println(temp(i))
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment