Spark大数据比赛经验记录(含sparksubmit 提交方法)
十月底, 参加了公司的一个spark大数据比赛, 题目比较简单, 但是由于自己缺乏此方面的业务知识, 所以对我来说解答的过程还是很有收获的 , 现在记录如下:
题目
数据表:
CREATE EXTERNAL TABLE fact_ipp_flux_limit(
clttime timestamp,
clttimeint bigint,
imsi string,
ci bigint,
url string,
tcpwrldelay bigint
)
PARTITIONED BY (
reportdate string,
reporthour int,
reportneid int)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ‘,’
STORED AS TEXTFILE LOCATION ‘/zxvmax/telecom/cn/test_data/fact_ipp_flux_limit’;
该数据表是统计的某地区电信用户手机上网情况, 其中clttime是访问对应url的时间, imsi是手机卡标识, ci是小区ID, url则是访问链接. 数据如下图所示:
问题 :
1. 根据提供的数据，输出每五分钟，人流量最大的TOP3小区。(简单)
时间：14：00 – 17：00
字段：时间(clttime),小区(ci), 用户(imsi)
2. 根据URL识别搜索引擎的关键词，并给出搜索热度排名。（中等）
时间：00:00 – 23:59
字段：url，时间(clttime)
3. 根据URL识别什么时段大家更喜欢看新闻。（中等）
时间：00:00 – 23:59
字段：url，时间(clttime)
4. 根据URL识别最受欢迎的网站。（中等）
时间：00:00 – 23:59
字段：url，时间(clttime)**
提交指南
我们以wordcount为例介绍下在linux环境下的提交方式
示例代码(wordcount.scala):
package bigdata
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object WordCount
{
def main(args: Array[String])
{
if (args.length < 1)
{
System.err.println("Usage: <file>")
System.exit(1)
}
val conf = new SparkConf()
val sc = new SparkContext(conf)
val line = sc.textFile(args(0))
val array = line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
array.saveAsTextFile("/odpp/files/output")
sc.stop()
}
}编译
scalac -classpath /home/mr/spark/lib/spark-assembly-1.4.1-hadoop2.5.0-cdh5.3.2.jar -d wordcount.jar wordcount.scala
此处要注意scala的版本要与spark-assembly的scala编译版本最好一致, 不然可能在运行时出错.上述jar是由scala 2.10.4编译的
提交
spark-submit --master yarn-cluster --queue user07_space --driver-memory 3G --executor-memory 2G --num-executors 3 --class bigdata.wordcount /home/wordcount.jar xxx.txt
其中,(./spark-submit --help可以查看详情)
yarn-cluster意思是放到yarn集群中执行作业, 如果单机模式的话可以用--master local[8]
--queue gzwspace 指定作业执行的用户空间，
--driver-memory 3G 指定启动作业时需要的内存大小
--executor-memory 2G 指定执行作业时，每个集群节点的内存大小
--num-executors 3 执行执行的节点数
--class bigdata.wordcount 指定class名
/home/wordcount.jar 指定jar包
xxx.txt 指定args参数–master参数的补充说明:
Master URL Meaning local 本地单线程 local[K] 本地K个线程, 建议设置城与cpu核数想相同 local[*] 本地多线程运行, 运行时的线程与cpu核数相同 spark://HOST:PORT 链接Spark Standlone集群管理器主机,默认端口为7077 yarn-client 采用client模式链接Yarn资源管理器, 集群位置信息配置在HADOOP_CONF_DIR或 YARN_CONF_DIR变量中 yarn-cluster 采用CLuster模式链接Yarn资源管理器, 集群位置信息与上相同 simr://HOST:PORT 兼容Hadoop 1.0 mesos://HOST:PORT 链接Mesos资源管理器
用到的点
- 使用spark-shell时候, 可以直接使用sqlContext提交sql语句, 比如:
sqlContext.sql(“select * from fact_ipp_flux_limit limit 10”).show
show是查看, 或者也可以打印: .collect.foreach(println) 但是使用spark-submit提交jar作业的时候, 要自己定义sqlContext:
val conf = new SparkConf()
val sc = new SparkContext(conf)
val sqlContext = new HiveContext(sc)shuffle时候task数目优化(经验值是可用cpu核数的三倍)
val buckets = try { sc.getConf.get("spark.cores.max").toInt * 3 } catch { case e: Exception => 200 } sqlContext.sql(s"set spark.sql.shuffle.partitions=$buckets")粗略的解释就是:如果这个值太低, 也就是task任务数太少, 使得每隔task处理时间过长容易Straggle, 而且有GC. 如果太大就有很多任务启动开销. 虽然设置大一点可以解决GC, 但有时候大了也没用是因为有数据倾斜(某一个task处理了大量任务,而其他task是清闲的)
url解码, 使用java的URLDecoder
try { result = java.net.URLDecoder.decode("%e4%bd%a0%e5%a5%bd", ENCODE);//encode可有可无,代表编码 } catch (UnsupportedEncodingException e) { e.printStackTrace(); }用零宽断言来提取搜索引擎词汇,如 百度的链接:
"""(?<=&word=).*(?=(&|$))""".r findFirstIn """fdsfsd&word=fdsfsdfsfdsf"""第一题里, 每五分钟, 可以将时间换算成分钟然后 /5, 数值相同的即是在同一个五分钟内
- 用开窗函数取每个分组的top3值:
row_number() over (partition by hour,minute5 order by usercount desc) - spark里的dataframe相关api要多学一下比如show, select等都很有用.
结论(简单版)
- 题目一
- 题目二
- 题目三
- 题目四
源代码
// 此次比赛提供的spark-assembly-1.4.1-hadoop2.5.0-cdh5.3.2版本是由scala2.10.4编译的
// 但是环境上scala版本为2.11.8, 执行会报错, 所以编译时请选择2.10.4的环境
package bigdata
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import scala.util.Try
object SparkGame {
def main(args: Array[String]) {
if (args.length < 1) {
System.err.println("============请在提交时添加参数: 1,2,3,4分别代表四个题目================")
System.exit(1)
}
val conf = new SparkConf()
val sc = new SparkContext(conf)
val sqlContext = new HiveContext(sc)
val buckets = try {
sc.getConf.get("spark.cores.max").toInt * 3
} catch {
case e: Exception => 200
}
sqlContext.sql(s"set spark.sql.shuffle.partitions=$buckets")
try {
args(0) match {
case "1" => top3Cell(sqlContext)
case "2" => searchHot(sqlContext)
case "3" => hotNews(sqlContext)
case "4" => popularWebSite(sqlContext)
case _ =>
System.err.println("=========参数错啦=========")
System.exit(1)
}
} catch {
case e: Exception => e.printStackTrace()
} finally {
sc.stop()
}
}
// 第一题 各时段Top3小区
private def top3Cell(sqlContext: HiveContext): Unit = {
def sql =
"""
select ci,usercount,date,hour,minute5*5 as minute
from (
select ci,date,hour,minute5,usercount,row_number() over (partition by hour,minute5 order by usercount desc) as rn
from(
select date,count(distinct imsi) as usercount,ci,hour,minute5
from (
select reportdate as date,reporthour as hour,cast(minute(clttime)/5 as int) as minute5,ci,imsi
from fact_ipp_flux_limit
where reportdate='2016-05-28' and reporthour in (14,15,16) and length(ci)=15
) a
group by hour,minute5,ci,date
) b
) c where rn<=3
"""
val temp = sqlContext.sql(sql).rdd.persist
temp.collect.foreach(println)
}
// 第二题 搜索引擎关键词排名
private def searchHot(sqlContext: HiveContext): Unit = {
def sql =
"""
select url from fact_ipp_flux_limit
"""
val baiduReg = """(?<=&word=).*(?=(&|$))""".r
val shenmaReg = """(?<=s\?q=).*(?=(&|$))""".r
val sogouReg = """(?<=keyword=).*(?=&)""".r
val q360Reg = """(?<=index.php\?q=).*(?=&)""".r
val searchKW = sqlContext.sql(sql).rdd.map(row => row.getString(0)).filter(_.length > 1)
.map { url =>
val baidu = baiduReg findFirstIn url
val shenma = shenmaReg findFirstIn url
val sogou = sogouReg findFirstIn url
val q360 = q360Reg findFirstIn url
val list = List(baidu, shenma, sogou, q360).filter(_ != None)
if (list.isEmpty)
""
else
Try {
list.head.getOrElse("").split("&").head
}.getOrElse("")
}.filter(_.length > 0)
.map { x => Try(java.net.URLDecoder.decode(x, "utf-8")).getOrElse("") }
.filter(_.length > 0)
.map((_, 1)).reduceByKey(_ + _)
.sortBy(_._2, ascending = false).persist
searchKW.take(150).foreach(println)
}
// 第三题 新闻热度
private def hotNews(sqlContext: HiveContext): Unit = {
import sqlContext.implicits._
def sql =
"""
select imsi,url,reporthour from fact_ipp_flux_limit
"""
val newsWebs = List("info.3g.qq.com", "kb.qq.com", "sina.cn", "toutiao.com", "yidianzixun", "ifeng.com", "sohu.com", "3g.163.com", "xinhuanet", "people.com", "myzaker", "weibo.com","news")
val hotNews = sqlContext.sql(sql).rdd.map(row => (row.getString(0), row.getString(1), row.getInt(2))).filter(_._2.length > 1)
.map { urlInfo => if (newsWebs.exists(urlInfo._2.contains(_))) URLInfo(urlInfo._1, "1", urlInfo._3) else URLInfo(urlInfo._1, "0", urlInfo._3) }
hotNews.toDF.registerTempTable("newsurl")
def sql2 =
"""select sum(p) as hotrate,reporthour
from (
select count(case when url=1 then 1 else null end)/count(*) as p,imsi,reporthour
from newsurl
group by imsi,reporthour
) as userrate
group by reporthour"""
sqlContext.sql(sql2).show()
}
// 第四题
private def popularWebSite(sqlContext: HiveContext): Unit = {
def sql =
"""
select url from fact_ipp_flux_limit
"""
val popularRDD = sqlContext.sql(sql).rdd
.map(row => row.getString(0))
.filter(_.length > 1)
.map(x => filterDomin(x))
.filter(_.length > 1)
.map(x => x.split("\\.").init.last + "." + x.split("\\.").last)
.map((_, 1)).reduceByKey(_ + _)
.sortBy(_._2, ascending = false).persist
//popularRDD.saveAsTextFile("/odpp/spaces/user07_space/files/popularwebsite")
popularRDD.take(150).foreach(println)
}
private def filterDomin(str: String): String = {
str.split("/", -1).dropRight(1).find(x => findTopDomain(x)).getOrElse("")
}
private def findTopDomain(x: String): Boolean = {
val domain = List(".cn", ".com", ".net", ".org", ".gov", ".edu")
domain.exists(x.contains(_))
}
}
case class URLInfo(imsi: String, url: String, reporthour: Int)
腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。
更多推荐
4
0- 0
已为社区贡献2条内容
所有评论(0)