模拟生成数据脚本

/*** 向文件中生产数据*/
object ProducePvAndUvData {//ipval IP = 223//地址val ADDRESS = Array("北京", "天津", "上海", "重庆", "河北", "辽宁","山西","吉林", "江苏", "浙江", "黑龙江", "安徽", "福建", "江西","山东", "河南", "湖北", "湖南", "广东", "海南", "四川","贵州", "云南", "山西", "甘肃", "青海", "台湾", "内蒙","广西", "西藏", "宁夏", "新疆", "香港", "澳门")//日期val DATE = new SimpleDateFormat("yyyy-MM-dd").format(new Date())//timestampval TIMESTAMP = 0L//useridval USERID = 0L//网站val WEBSITE = Array("www.baidu.com", "www.taobao.com", "www.dangdang.com", "www.jd.com", "www.suning.com", "www.mi.com", "www.gome.com.cn")//行为val ACTION = Array("Regist", "Comment", "View", "Login", "Buy", "Click", "Logout")def main(args: Array[String]): Unit = {val pathFileName = "G://idea//scala//spark02/data"//创建文件val createFile = CreateFile(pathFileName)//向文件中写入数据 需要的对象val file = new File(pathFileName)val fos = new FileOutputStream(file, true)val osw = new OutputStreamWriter(fos, "UTF-8")val pw = new PrintWriter(osw)if (createFile) {var i = 0//产生5万+数据while (i < 50000){ //模拟一个ipval random = new Random()val ip = random.nextInt(IP) + "." + random.nextInt(IP) + "." + random.nextInt(IP) + "." + random.nextInt(IP)//模拟地址val address = ADDRESS(random.nextInt(34))//模拟日期val date = DATE//模拟useridval userid = Math.abs(random.nextLong)/*** 这里的while模拟是同一个用户不同时间点对不同网站的操作*/var j = 0var timestamp = 0Lvar webSite = "未知网站"var action = "未知行为"val flag = random.nextInt(5) | 1while (j < flag) { //                   Threads.sleep(5);//模拟timestamptimestamp = new Date().getTime()//模拟网站webSite = WEBSITE(random.nextInt(7))//模拟行为action = ACTION(random.nextInt(6))j += 1/*** 拼装*/val content = ip + "\t" + address + "\t" + date + "\t" + timestamp + "\t" + userid + "\t" + webSite + "\t" + actionSystem.out.println(content)//向文件中写入数据pw.write(content + "\n")}i += 1}//注意关闭的先后顺序，先打开的后关闭，后打开的先关闭pw.close()osw.close()fos.close()}}/*** 创建文件*/def CreateFile(pathFileName: String): Boolean = {val file = new File(pathFileName)if (file.exists) file.deleteval createNewFile = file.createNewFile()System.out.println("create file " + pathFileName + " success!")createNewFile}
}

统计每个网站的PU、VU、每个网站的每个地区访问量，由大到小排序

 def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setMaster("local")conf.setAppName("SparkPvAndUv")val sc = new SparkContext(conf)val rdd: RDD[String] = sc.textFile("G:/idea/scala/spark02/data")println("*************PU******************")rdd.map(line=>{(line.split("\t")(5),1)}).reduceByKey(_+_).sortBy(_._2,false)//是否降序，false:是降序.foreach(println)println("*************UV******************")rdd.map(line=>line.split("\t")(5)+"_"+line.split("\t")(1))//网站_ip.distinct()//去重.map(line=>{(line.split("_")(0),1)}).reduceByKey(_+_).sortBy(_._2,false).foreach(println)//每个网址的每个地区访问量，由大到小排序val site_local: RDD[(String, String)] = rdd.map(line=>{(line.split("\t")(5),line.split("\t")(1))})val site_localIterable: RDD[(String, Iterable[String])] = site_local.groupByKey()val result: RDD[(String, AbstractSeq[(String, Int)])] = site_localIterable.map(one => {val localMap = mutable.Map[String, Int]()//可变mapval site = one._1val localIterator = one._2.iteratorwhile (localIterator.hasNext) {//地区val local = localIterator.next()if (localMap.contains(local)) {//如果map中有该地区，则获取该地区的值再加1val value = localMap.get(local).getlocalMap.put(local, value + 1)} else {//如果map中没有该地区，则获取该地区的值再加1localMap.put(local, 1);}}//默认是升序，降序：localMap.toList.sortBy(-_._2)，既多一个"-"val tuples: List[(String, Int)] = localMap.toList.sortBy(-_._2)if (tuples.length > 3) {val list = new ListBuffer[(String, Int)]()for (i <- 0 to 2) {list.append(tuples(i))}(site, list)} else {(site, tuples)}})result.foreach(println)}