C/C++教程
sparksql和spark core 实现地域报表分析
本文主要是介绍sparksql和spark core 实现地域报表分析，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
文章目录
- 计算逻辑
- spark sql实现代码
- spark core实现代码
- 读取的普通日志文件
- 读取parquet文件
计算逻辑
spark sql实现代码
package com.dmp.report
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
object AreaAnalysRpt {
def main(args: Array[String]): Unit = {
// 0 校验参数个数
if (args.length != 1) {
println(
"""
|cn.dmp.report.AreaAnalyseRpt
|参数：
| logInputPath
""".stripMargin)
sys.exit()
}
// 1 接受程序参数
val Array(logInputPath) = args
// 2 创建sparkconf->sparkContext
val sparkConf = new SparkConf()
sparkConf.setAppName(s"${this.getClass.getSimpleName}")
sparkConf.setMaster("local[*]")
// RDD 序列化到磁盘 worker与worker之间的数据传输
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val sc = new SparkContext(sparkConf)
val sQLContext = new SQLContext(sc)
// 读取parquet文件
val parquetData: DataFrame = sQLContext.read.parquet(logInputPath)
// dataframe -> table
parquetData.registerTempTable("log")
//业务逻辑
sQLContext.sql(
"""
|select
|provincename, cityname,
|sum(case when requestmode=1 and processnode >=2 then 1 else 0 end) 有效请求,
|sum(case when requestmode=1 and processnode =3 then 1 else 0 end) 广告请求,
|sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid !=0 then 1 else 0 end) 参与竞价数,
|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) 竞价成功数,
|sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) 展示数,
|sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) 点击数,
|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*adpayment/1000 else 0 end) 广告成本,
|sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*winprice/1000 else 0 end) 广告消费
|from log
|group by provincename, cityname
""".stripMargin
).show()
}
}spark core实现代码
处理计算指标的逻辑类
package cn.dmp.utils
object RptUtils {
/**
* List(原始请求，有效请求，广告请求)
*/
def caculateReq(reqMode: Int, prcNode: Int): List[Double] = {
if (reqMode == 1 && prcNode == 1) {
List[Double](1, 0, 0)
} else if (reqMode == 1 && prcNode == 2) {
List[Double](1, 1, 0)
} else if (reqMode == 1 && prcNode == 3) {
List[Double](1, 1, 1)
} else List[Double](0, 0, 0)
}
/**
* List(参与竞价，竞价成功，消费，成本)
*/
def caculateRtb(effTive: Int, bill: Int, bid: Int, orderId: Int, win: Int, winPrice: Double, adPayMent: Double): List[Double] = {
if (effTive == 1 && bill == 1 && bid == 1 && orderId != 0) {
List[Double](1, 0, 0, 0)
} else if (effTive == 1 && bill == 1 && win == 1) {
List[Double](0, 1, winPrice / 1000.0, adPayMent / 1000.0)
} else List[Double](0, 0, 0, 0)
}
/**
* List(广告展示，点击)
*/
def caculateShowClick(reqMode: Int, effTive: Int): List[Double] = {
if (reqMode == 2 && effTive == 1) {
List[Double](1, 0)
} else if (reqMode == 3 && effTive == 1) {
List[Double](0, 1)
} else List[Double](0, 0)
}
}读取的普通日志文件
package com.dmp.report
import com.dmp.beans.Log
import com.dmp.utils.RptUtils
import org.apache.spark.{SparkConf, SparkContext}
/*
F:\牛牛学堂大数据24期\09-实训实战-9天\dmp&&移动项目\dmp\2016-10-01_06_p1_invalid.1475274123982.log.FINISH.bz2
C:\Users\admin\Desktop\result3
*
*/
object AreaAnalysRpt2 {
def main(args: Array[String]): Unit = {
// 0 校验参数个数
if (args.length != 2) {
println(
"""
|cn.dmp.report.AreaAnalyseRpt
|参数：
| logInputPath
| resultOutputPath
""".stripMargin)
sys.exit()
}
// 1 接受程序参数
val Array(logInputPath, resultOutputPath) = args
// 2 创建sparkconf->sparkContext
val sparkConf = new SparkConf()
sparkConf.setAppName(s"${this.getClass.getSimpleName}")
sparkConf.setMaster("local[*]")
// RDD 序列化到磁盘 worker与worker之间的数据传输
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val sc = new SparkContext(sparkConf)
sc.textFile(logInputPath)
.map(_.split(",", -1))
.filter(_.length >= 85)
.map(arr => {
val log = Log(arr)
val req = RptUtils.caculateReq(log.requestmode, log.processnode)
val rtb = RptUtils.caculateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin, log.winprice, log.adpayment)
val showClick = RptUtils.caculateShowClick(log.requestmode, log.iseffective)
// (省，地市，媒体，渠道，操作系统，网络类型,...，List(9个指标数据))
((log.provincename, log.cityname), req ++ rtb ++ showClick)
}).reduceByKey((list1, list2) => {
//将两个list拉链组合在一个,相同位置的元素组成一个tuple,map操作进行累加
list1.zip(list2).map(t => t._1 + t._2)
})// 会打印一个(),使用map将数据进行整理
.map(t => t._1._1 + ","+t._1._2+","+t._2.mkString)
.saveAsTextFile(resultOutputPath)
}
}读取parquet文件
package com.dmp.report
import com.dmp.utils.RptUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
/*
C:\Users\admin\Desktop\result1\
C:\Users\admin\Desktop\result3
*
*/
object AreaAnalysRpt3 {
def main(args: Array[String]): Unit = {
// 0 校验参数个数
if (args.length != 2) {
println(
"""
|cn.dmp.report.AreaAnalyseRpt
|参数：
| logInputPath
| resultOutputPath
""".stripMargin)
sys.exit()
}
// 1 接受程序参数
val Array(logInputPath, resultOutputPath) = args
// 2 创建sparkconf->sparkContext
val sparkConf = new SparkConf()
sparkConf.setAppName(s"${this.getClass.getSimpleName}")
sparkConf.setMaster("local[*]")
// RDD 序列化到磁盘 worker与worker之间的数据传输
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val sc = new SparkContext(sparkConf)
val sQLContext = new SQLContext(sc)
val parquetData = sQLContext.read.parquet(logInputPath)
parquetData.map(row => {
//从row中获取具体字段
// 是不是原始请求，有效请求，广告请求 List(原始请求，有效请求，广告请求)
val reqMode = row.getAs[Int]("requestmode")
val prcNode = row.getAs[Int]("processnode")
// 参与竞价, 竞价成功 List(参与竞价，竞价成功, 消费, 成本)
val effTive = row.getAs[Int]("iseffective")
val bill = row.getAs[Int]("isbilling")
val bid = row.getAs[Int]("isbid")
val orderId = row.getAs[Int]("adorderid")
val win = row.getAs[Int]("iswin")
val winPrice = row.getAs[Double]("winprice")
val adPayMent = row.getAs[Double]("adpayment")
// 转换为list
val reqList = RptUtils.caculateReq(reqMode, prcNode)
val rtbList = RptUtils.caculateRtb(effTive, bill, bid, orderId, win, winPrice, adPayMent)
val showClickList = RptUtils.caculateShowClick(reqMode, effTive)
//返回元组
// 返回元组
((row.getAs[String]("provincename"), row.getAs[String]("cityname")), reqList ++ rtbList ++ showClickList)
}).reduceByKey((list1, list2) => {
list1.zip(list2).map(t => t._1 + t._2)
}).map(t => t._1._1 + "," + t._1._2 + "," +t._2.mkString(","))
.saveAsTextFile(resultOutputPath)
sc.stop()
}
}
这篇关于sparksql和spark core 实现地域报表分析的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
增量更新怎么做？-icode9专业技术文章分享11-23
-
压缩包加密方案有哪些？-icode9专业技术文章分享11-23
-
用shell怎么写一个开机时自动同步远程仓库的代码？-icode9专业技术文章分享11-23
-
webman可以同步自己的仓库吗？-icode9专业技术文章分享11-23
-
在 Webman 中怎么判断是否有某命令进程正在运行？-icode9专业技术文章分享11-23
-
如何重置new Swiper?-icode9专业技术文章分享11-23
-
oss直传有什么好处？-icode9专业技术文章分享11-23
-
如何将oss直传封装成一个组件在其他页面调用时都可以使用？-icode9专业技术文章分享11-23
-
怎么使用laravel 11在代码里获取路由列表？-icode9专业技术文章分享11-23
-
怎么实现ansible playbook 备份代码中命名包含时间戳功能？-icode9专业技术文章分享11-22
-
ansible 的archive 参数是什么意思？-icode9专业技术文章分享11-22
-
ansible 中怎么只用archive 排除某个目录？-icode9专业技术文章分享11-22
-
exclude_path参数是什么作用？-icode9专业技术文章分享11-22
-
微信开放平台第三方平台什么时候调用数据预拉取和数据周期性更新接口？-icode9专业技术文章分享11-22
-
uniapp 实现聊天消息会话的列表功能怎么实现？-icode9专业技术文章分享11-22
栏目导航
