sparkSQL实例_flink sql

sparkSQL实例_flink sql记一次SparkSqlETL过程需求:1)input:json日志2)ETL:根据IP解析出省份,城市3)stat:地区分布指标计算,满足条件的才算,满足条件的赋值为1,不满足的赋值为0(如下图)将统计结果写入MySQL中。(就比如说这个广告请求要满足requestmode=1和processnode=3这两个条件)valspark=SparkSessi…

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全家桶1年46,售后保障稳定

记一次SparkSql ETL 过程

需求说明

1)input:json日志
2)ETL:根据IP解析出 省份,城市
3)stat: 地区分布指标计算,
满足条件的才算,满足条件的赋值为1,不满足的赋值为0 (如下图)
将统计结果写入MySQL中。
(就比如说这个广告请求要满足 requestmode=1 和 processnode =3 这两个条件)在这里插入图片描述

代码分析

val spark = SparkSession.builder().master("local[2]").appName("LogApp").getOrCreate()
import spark.implicits._
val inputDF = spark.read.json("inputdata/data-test.json")
inputDF.printSchema()
// ETL: 一定保留原有的数据 最完整 而且要落地 (理由:要是数据出错好重新计算)
val newDF = inputDF.withColumn("province", MyUDF.getProvince(inputDF.col("ip")))
.withColumn("city", MyUDF.getCity($"ip"))//自定义udf 函数
.write.format("parquet")
.mode(SaveMode.Overwrite)
.save("outparquet") // 最好保存parquet格式 (spark默认就是parquet + snappy)
// 计算 重新去读取etl之后的数据源 
val parquetDF = spark.read.parquet("outparquet/xxx.snappy.parquet")
parquetDF.printSchema()
parquetDF.show(5)
parquetDF.createOrReplaceTempView("log")
//业务SQL
val areaSQL01 = "select province,city, " +
"sum(case when requestmode=1 and processnode >=1 then 1 else 0 end) origin_request," +
"sum(case when requestmode=1 and processnode >=2 then 1 else 0 end) valid_request," +
"sum(case when requestmode=1 and processnode =3 then 1 else 0 end) ad_request," +
"sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) bid_cnt," +
"sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) bid_success_cnt," +
"sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) ad_display_cnt," +
"sum(case when requestmode=3 and processnode=1 then 1 else 0 end) ad_click_cnt," +
"sum(case when requestmode=2 and iseffective=1 and isbilling=1 then 1 else 0 end) medium_display_cnt," +
"sum(case when requestmode=3 and iseffective=1 and isbilling=1 then 1 else 0 end) medium_click_cnt," +
"sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>20000 then 1*winprice/1000 else 0 end) ad_consumption," +
"sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>20000 then 1*adpayment/1000 else 0 end) ad_cost " +
"from log group by province,city"
spark.sql(areaSQL01).createOrReplaceTempView("area_tmp")
val areaSQL02 = "select province,city, " +
"origin_request," +
"valid_request," +
"ad_request," +
"bid_cnt," +
"bid_success_cnt," +
"bid_success_cnt/bid_cnt bid_success_rate," +
"ad_display_cnt," +
"ad_click_cnt," +
"ad_click_cnt/ad_display_cnt ad_click_rate," +
"ad_consumption," +
"ad_cost from area_tmp " +
"where bid_cnt!=0 and ad_display_cnt!=0"
// 写入MySQL (上一篇博客有介绍)
val config = ConfigFactory.load()
val url = config.getString("db.default.url")
val user = config.getString("db.default.user")
val password = config.getString("db.default.password")   
spark.sql(areaSQL02)
.write.format("jdbc")
.option("url", url)
.option("dbtable", "sparksql_test")
.option("user", user)
.option("password", password)
.mode(SaveMode.Overwrite)
.save()
spark.stop()

Jetbrains全家桶1年46,售后保障稳定

自定义udf 函数代码

object MyUDF { 

import org.apache.spark.sql.functions._
def getProvince = udf((ip:String)=>{ 

val cityInfo = IPUtil.getCityInfo(ip)
val splits = cityInfo.split("\\|")
var city = "未知"
if (splits.length == 5){ 

city = splits(2)
}
city
})
def getCity = udf((ip:String)=>{ 

val cityInfo = IPUtil.getCityInfo(ip)
val splits = cityInfo.split("\\|")
var city = "未知"
if (splits.length == 5){ 

city = splits(3)
}
city
})
}

调优

① ETL 落地过程中应该调用coalesce() 防止产生多个小文件

 val newDF = inputDF.withColumn("province", MyUDF.getProvince(inputDF.col("ip")))
.withColumn("city", MyUDF.getCity($"ip"))
.coalesce(1)
.write.format("parquet").mode(SaveMode.Overwrite).save("outparquet")

② spark.conf.set(“spark.sql.shuffle.partitions”,“400”) 修改SparkSql shuffle task数量,默认是200
在这里插入图片描述

总结

ETL过程:
input:json
清洗 ==> ODS 大宽表 HDFS/Hive/SparkSQL
output: 列式存储 ORC/Parquet (列式存储) (为啥要用这两种? 因为ETL清洗出来的是全字段,我们不可能使用到全部字段,所以采用列式存储,用到几列就获取几列,这样就能减少I/O,性能大大提升)

Stat
==> 一个非常简单的SQL搞定
==> 复杂:多个SQL 或者 一个复杂SQL搞定

列式:ORC/Parquet
特点:把每一列的数据存放在一起
优点:减少IO 需要哪几列就直接获取哪几列
缺点:如果你还是要获取每一行中的所有列,那么性能比行式的差

行式:MySQL
一条记录有多个列 一行数据是存储在一起的
优点:
你每次查询都使用到所有的列
缺点:
大宽表有N多列,但是我们仅仅使用其中几列

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/223422.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • Docker常用命令

    Docker常用命令

  • APP抓包工具Fidder

    APP抓包工具FidderAPP抓包工具Fidder0.安装下载Classic版本https://www.telerik.com/download/fiddler1.使用a.设置Fiddler打开Fiddler,Tools->FiddlerOptions(配置完后记得要重启Fiddler)选中”DecrptHTTPStraffic”,Fiddler就可以截获HTTPS请求选中”Allowremotecomputerstoconnect”.是允许别的机器把HTTP/HTTPS请求

  • 浏览器怎么打开微信客户端连接服务器,微信“请在微信客户端打开链接”怎么办?-在浏览器中打开微信链接的方法 – 河东软件园…「建议收藏」

    浏览器怎么打开微信客户端连接服务器,微信“请在微信客户端打开链接”怎么办?-在浏览器中打开微信链接的方法 – 河东软件园…「建议收藏」自从出现了电脑版的微信之后,很多用户都会在电脑中下载安装一个客户端,可就是电脑客户端中打开链接也会出错!微信中有的时候朋友或是公众号会发送一些链接,若是使用电脑单击打开就会被提示“请在微信客户端打开链接”,可是自己使用的就是电脑客户端,并且更换浏览器也不能解决这个现象,这是怎么一回事呢?因为在微信中是自动设置了使用默认浏览器打开的,无法识别的时候自然就不能打开了,我们可以在微信中直接将这个功能关闭…

  • asp:UpdatePanel客户端回传事件管理

    asp:UpdatePanel客户端回传事件管理asp:UpdatePanel客户端回传事件管理Asp:UpdatePanel是在Asp.NetWebForm中的一个局部刷新控件,虽然很好用,但是在使用过程中却发现如果局部刷新的数据需要再次使用页面js进行格式化,页面则会乱套,所以在这里我们需要对UpdatePanel的回传过程进行控制。

  • linux安装配置Mysql详细步骤

    linux安装配置Mysql详细步骤下载Mysqlrpm包官网下载:https://dev.mysql.com/downloads/mysql/如果你到了这里,本地应该有了如下四个rpm包:mysql-community-client-5.7.26-1.el6.x86_64.rpmmysql-community-common-5.7.26-1.el6.x86_64.rpmmysql-community-libs-5…

  • 信息学奥赛一本通(C++版) 网站补充题目

    信息学奥赛一本通(C++版) 网站补充题目总目录详见:https://blog.csdn.net/mrcrack/article/details/86501716信息学奥赛一本通(C++版)网站补充题目http://ybt.ssoier.cn:8088//1414【2017NOIP普及组】成绩//样例通过,提交AC2017-12-3121:45#include<stdio.h>intmain(){…

    2022年10月27日

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号