开发者学堂课程【2020版大数据实战项目之 DMP 广告系统（第五阶段）：报表统计_广告投放统计_代码实现】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/680/detail/11820
报表统计_广告投放统计_代码实现
一、代码编写
1、刚才已经简单梳理过需求了，在刚才的环节当中一定要把握住两个点，第一个点要统计的是广告相关的指标，按照省市的信息作为维度来统计这些指标，通过本课程进行具体的代码编写，把报表写完。
2、在 report 目录下新建一个 scala class，scala class 叫做 AdsRegionReportProcessor。把 class 改为 object，就能继续继承 possessor，是 ReportPossess 。下面写第一个方法，要读的源表的表名叫 ODS 表，就可以使用 ETLRunner.ODS_TABLE_NAME 。接下来要进行具体的处理，具体处理的时候，这种需求要想使用 dataframe 命令式的 ITI 可能是不太明智的，因为这个需求稍微有点复杂，所以先写一个 SQL 语句，用 SQL 语句进行相应的编写。使用一个 SQL 语句统计指标，把它命名为 KPI，KPI 就是指标的意思，命名为 kpiSQL。先统计指标，统计指标以后还要再进行一个小小的统计，就是按照指标里面的一些指示去统计比例。比例的统计叫做 rateSQL，分为两个 SQL 编写，先写 kpiSQL，第一条 SQL 就三个引号，按回车，IDEA 会自动的补全。我们要统计的是广告运行的广告投放的指标在省和市的分布情况。只要说在什么的统计情况，就直接 select 一个东西，先放在那里，from 不知道，也先空着。只要按照什么统计，一定有 group by，group by 一个叫 region，一个叫 city，group by 以后后面或前面不知道还有没有 val，先把 SQL 放在这。如果 group by了 region 和 city，select 后面一定要添加 region和 city，这是约定俗成的。按照 region 和 city进行 group by，前面必不可少的一定要进行一个 region 和 city 的统计。from 什么不知道，前面命名一个临时表的名字 tempTable，tempTable 可以叫做 sourceTableName，sourceTableName 可以叫做 ODS_table。在这个字符串前面加一个 s，加了 s 以后 from $ sourceTableName，这样整体上 sql 语句的框架就搭建出来了，搭出来以后继续往下写。
3、先统计第一个指标，第一个指标要统计所有请求的数量，要做一个判断，如果统计的是数量的话，直接写一个 sum，因为统计的是数量，在什么维度上统计数量，怎么统计的这个数量，这有分组，这个 sum 就是 sum 分组内的东西。Sum 里面应该写什么？要统计符合有效请求数量规则的所有的数量，数量已经统计出来了，就是 sum，符合这个规则就算一。前面加一个 case，when 后面跟上条件的判断 ，第一个 requestmode = 1，并且 processnode ＞= 1，有效请求的数量大于等于一。如果符合这个 when就返回一个1，如果不符合就返回一个0，在后面加上一个 end。case、when、end、then 就是一个常见的句式，当这个条件满足的时候，这条数据就返回一个1，如果不满足这条数据，就返回一个0。这个 case、when、then 作用于数据集当中的每一条数据。
import cn. itcast. etl.ETLRunner
import org. apache. kudu. Schema
import org. apache. spark. sql.DataFrame
object AdsRegionReportProcessor extends ReportProcessor {
override def sourceTableName(): String={
ETLRunner.ODS_TABLE_NAME
}
override def process(dataFrame: DataFrame): DataFrame= {
val sourceTableName="ods_table"
val kpiSQL =
s"""
| select
| region, city,
| sum(case
| when(requestmode=1 and processnode>=1) then 1
| else 0
|end),
| from
| $sourceTableName
| group by
| region, city
""".stripMargin
4、接下来继续往下看，这是第一个，第二个统计有效的请求和广告请求数量，应该是 processnode＞=2。一般这种情况在取一个字段的时候，在取一条数据的一个字段的时候，应该使用别名，所以这个地方为它添加一个别名，这个别名就叫做 s。所以 when 后面应该是 s. requestmode=1 和 s. processnode>=1。复制 sum，sum 复制完以后也是 case when then，但是这个 case、when、then 大于等于二，这就是和刚才的区别。但是这两个 sum 在统计完以后并没有给它命名，所以要给他们简单的命下名。第一个应该 as 有效的请求数量，叫 original_req_cnt，第二个叫做 valid_req_cnt，这两个内容已经统计出来了，统计出来以后再去看第三个。第三个还是这个 sum，刚才是 process 大于等于二，有效的请求数量，接下来要让它等于三，现在统计的是广告的请求数量。还是满足这个条件给1，不满足给0，这个应该命名为 ad_req_cnt，第三个指标也已经统计出来了。
val kpiSQL =
s"""
| select
| region, city,
|sum(case
|when(s.requestmode=1 and s.processnode>=1) then 1
| else 0
|end), as original_req_cnt
|sum(case
| when(s.requestmode=1 and s.processnode>=2) then 1
|else 0
|end), as valid_req_cnt
|sum(case
|when(s.requestmode=1 and s.processnode=3) then 1
|else 0
|end), as ad_req_cnt
| from
| $sourceTableName s
| group by
| region, city
""".stripMargin
5、下面要统计 AdExchange 的有效记录数量，AdExchange 就不在手写了，直接把 sql 语句拷贝过来。现在要统计的是一些指标，大家要了解 sql 的格式。从我已经写好的代码当中进行复制，
这个时候这个 sql 语句就完成了。这条 sql 语句大致就是一个又一个的 sum，整个结构大家能理解，具体怎么进行统计不是重点。
| sum(CASE
| WHEN (t. adplatformproviderid >=100000
|ANDt. iseffective=1
|ANDt. isbilling=1
|AND t. isbid=1
|ANDt. adorderid 1=0) THEN1
|ELSE 0
|END)AS join_rtx_cnt,
|sum(CASE
| WHEN (t. adplatformproviderid >=100000
|ANDt. iseffective=1
|ANDt. isbilling=1
|AND t. iswin = 1) THEN 1
|ELSE 0
|END)AS success_rtx_cnt,
|sum(CASE
|WHEN(t. requestmode=2
|ANDt. iseffective=1) THEN1
|ELSE 0
|END)AS ad_show_cnt,
|sum(CASE
|WHEN(t. requestmode=3
|ANDt. iseffective=1) THEN1
|ELSE 0
|END)AS ad_click_cnt,
|sum(CASE
|WHEN(t. requestmode=2
|ANDt. iseffective=1)
|ANDt. isbilling=1 THEN1
|ELSE 0
|END)AS media_show_cnt,
|sum(CASE
|WHEN(t. requestmode=3
|ANDt. iseffective=1)
|ANDt. isbilling=1 THEN1
|ELSE 0
|END)AS media_click_cnt,
| sum(CASE
|WHEN (t. adplatformproviderid >=100000
|ANDt. iseffective=1
|ANDt. isbilling=1
|AND t. iswin = 1)
|AND t. aborderid> 20000
|ANDt. adcreativeid>200000) THEN floor(t. adpayment/1000)
| ELSE 0
| END)AS dsp_pay_money,
| sum(CASE
| WHEN (t. adplatformproviderid >=100000
|ANDt. iseffective=1
|ANDt. isbilling=1
|AND t. iswin = 1)
|AND t. aborderid> 20000
|ANDt. adcreativeid>200000) THEN floor(t. adpayment/1000)
|ELSE 0
| from
| $sourceTableName s
| group by
| region, city
""".stripMargin
6、首先，这个数据没有那么准确，跟大家以后工作当中的数据可能不太一样；其次，大家无需关心这个数据的细节，只需要了解整体的步骤，sql 语句的格式，主要是了解格式。这个 sql 语句已经表示出来了，要针对这个 SQL 语句结果再进行一次统计，
要统计一些占比。占比大概就是这样一条 sql 语句，
lazy val sqlWithRate =
s" " "
|SELECT t.*,
| t. success rtx cnt/t. join rtx cnt AS success_rtx_rate,
| t. media click cnt/t. ad click cnt AS ad_click_rate
|FROM$MID_TEMP_TABLEt
|WHEREt. success_rtx_cnt!=0
|ANDt. Join_rtx_cnt!=0
|ANDt. Media_click_cnt!=0
|ANDt. ad_click_cnt!=0
""" .stripMargin
有一个 WHERE，FROM 的是一张临时表，这张临时表就是刚才如果在进行 sql 语句执行以后执行的结果。在这个结果的基础之上进行一次 where，判断 success_rtx_cnt 不等于零，join_rtx_cnt、media_click_cnt、ad_click_cnt 都不等于零，也就是进行占比的统计当中，这四个被除数，就是这四个数，两个被除数是不能等于零的。
7、接下来进行一个相应的除法，前面的所有字段都要，后面是追加的两个字段，这就是这一段 SQL 语句直接拷贝放在代码里面，sql 没有必要自己去写，直接拷贝
大家了解是它是怎么做的就可以了，整个 sql 结构是怎么样的就可以。这个时候整个代码算是已经开发完成了，其实也没有。
val rateSQL
s" " "
|SELECT t.*,
|t. success rtx cnt/t. join rtx cnt AS success_rtx_rate,
|t. media click cnt/t. ad click cnt AS ad_click_rate
|FROM$MID_TEMP_TABLEt
|WHEREt. success_rtx_cnt!=0
|ANDt. Join_rtx_cnt!=0
|ANDt. Media_click_cnt!=0
|ANDt. ad_click_cnt!=0
""" .stripMargin
再去命名一个中间的名字，叫做 kpiTableName，kpi_table，
把 kpiTableName 复制。
8、接下来执行这些 SQL 语句，首先从外面拿到一个 dataFrame，拿到这个dataFrame 以后要 createOrReplaceTempView，第一个要注册的邻表是给第一个 sql 语句使用的，这个邻表叫做 sourceTableName，这个时候返回了一个 source 的结果，其实就是 kpi 的一个结果，kpiTable 统计过以后，执行这个语句，dataFrame.sparkSession.sql（），把 kpiSQL 放进去，执行一个 sql 才能获取到相应的结果，而这个 kpiSQL 就是第一条很长的 SQL。接下来获取到的结果是 kpiTable，获取到 kpiTable 以后，拿 kpiTable createOrReplaceTempView，把它命名为 kpiTableName。可以从 kpiTableName 中读取数据执行这条 sql 语句。接下来 val 拿到一个新的结果，这个结果就是 rateTable，通过 kpiTable或者是从外面直接传的 dataFrame获取到 sparkSession.sql 也是可以的，把 readSQL 放进去，rateTable 就是这一次计算的结果。
dataFrame.createOrReplaceTempView(sourceTableName)
val kpiTable = dataFrame.sparkSession. sql(kpiSQL)
kpiTable.createOrReplaceTempView(kpiTableName)
val rateTable = dataFrame.sparkSession. sql(rateSQL)
rateTable
9、把结果返回回去，结果返回回去以后，要提供一个结果目标表的表名，表名把它命名为 report_ads_region，还是要加上日期，直接使用 KudoHelper，拿到其中叫做 formattedDate，表名已经准备好了，按照每两个 key 进行分区非常明确，就是一个 region，一个 city。接下来要把 schema 补全，首先，schema 里面先执行这条 sql 语句，
这条 SQL 语句执行完以后统计所有的列，这些列统计过以后，执行 rateSQL，在刚才的基础之上又增加了两列，一个叫做 success_rtx_rate，一个叫做 ad_click_rate，这是前面第一个结果集的所有的列，再加上后面的两列，总共就是这么多列，这么多列还要涉及数据类型之类的问题，所以 schema 就直接复制，知道里面都有什么东西就可以了，把他放在 targetTableSchema 当中。这个时候把 type 也导入，有一个叫做 KudoType，这个时候整个类就编写好了。
override def targetTableName(): String= {
“report_ads_region_ ”+KuduHelper.formattedDate()
}
override def targetTableSchema(): Schema = {
import scala.collection.JavaConverters._
override def targetTableKeys(): List[String] ={
List("region",“city”)
}
基于 process 这个框架写，只需要按部就班的把每一个方法都实现就可以了。
10、这个时候在 DailyReportRunner 当中注册 AdsRegionReportProcessor，这样整段代码就完成了，然后运行一下。这一小节就结束了，没有报任何错误，代码已经执行完了，执行完以后应该到 Kudo 当中查看这张报表有没有生成。进入到 Kudo 当中，打开 Tables，发现有一张表叫做 report_ads_region_190709
这张表已经创建出来了，创建出来以后复制一下再 impala 当中的键表语句，
CREATE EXTERNAL TABLE ` report_data_region 190709 ` STORED AS KUDU
TBLPROPERTIES(
'kudu. Table_name'='report_data_region_190709',
'kudu. master_ addresses'='cdh01:7051,cdh02;7051,
cdh03:7051')
进入 shell 窗口当中，impala shell，use dmp 这个数据库，然后创建这张表，输入分号，再回车，这张表已经创建出来了。select * from report_data_region_190709，然后运行 SQL 语句，
整个的数据就出来了，这些指标就都有了，
本课程到此讲解完毕。
