Spark Streaming实战对论坛网站动态行为pv,uv,注册人数,跳出率的多维度分析_小强签名设计 的博客-CSDN博客_spark streaming uv
实时统计每天pv,uv的sparkStreaming结合redis结果存入mysql供前端展示
实时统计每天pv,uv的sparkStreaming结合redis结果存入mysql供前端展示_大数据技术派-CSDN博客_spark streaming uv
Flume+Kafka+Storm+Redis构建大数据实时处理系统:实时统计网站PV、UV+展示
flume+kafka+slipstream实现黑名单用户访问实时监测
实战SparkStream+Kafka+Redis实时计算商品销售额实战SparkStream+Kafka+Redis实时计算商品销售额_小哈-whzhaochao-CSDN博客_sparkstream+kafka
spark streaming从kafka获取数据,计算处理后存储到redis
深入理解Spark Streaming流量控制及反压机制_lzf的博客-CSDN博客_spark streaming反压机制
大数据采集、清洗、处理:使用MapReduce进行离线数据分析完整案例大数据采集、清洗、处理:使用MapReduce进行离线数据分析完整案例_香飘叶子的技术博客_51CTO博客_mapreduce数据清洗案例大数据采集的方法
一、大数据处理的常用方法、项目的流程:
在互联网应用中,不管是哪一种处理方式,其基本的数据来源都是日志数据,例如对于web应用来说,则可能是用户的访问日志、用户的点击日志等。
大数据处理目前比较流行的是两种方法,一种是离线处理,一种是在线处理,基本处理架构如下:
当然,如果只是希望得到数据的分析结果,对处理的时间要求不严格,就可以采用离线处理的方式,比如我们可以先将日志数据采集到HDFS中,之后再进一步使用MapReduce、Hive等来对数据进行分析,这也是可行的。
- Flume将网站日志数据采集到HDFS分布式存储系统中
- Spark SQL清洗存储在HDFS的网站日志数据,清洗完后将其数据继续存储在HDFS中
- Hive建立数据仓库,建立外部表,将清洗完的日志数据从HDFS中导入到Hive的外部表中,作为基础数据的存储
- 在Hive中新建新的外部表用于存储PV、UV的结果数据
- 用Hive的HQL统计分析日志数据,统计出PV、UV并将结果数据存到新的外部表中
- 将统计完的PV、UV数据使用Sqoop从Hive同步到外部的MySQL中供给WEB前端使用
如果对于数据的分析结果在时间上有比较严格的要求,则可以采用在线处理的方式来对数据进行分析,如使用Spark、flink等进行处理。比较贴切的一个例子是天猫双十一的成交额,在其展板上,我们看到交易额是实时动态进行更新的,对于这种情况,则需要采用在线处理。
- 如何一步步构建我们的实时处理系统(Flume+Kafka+Storm+Redis)
- 1.Flume将网站日志数据采集到kafka、
- 2.sparkstreaming实时处理kafka数据网站的用户访问日志,并统计出该网站的PV、UV
- 3.将实时分析出的PV、UV等指标,实时处理后发送kafka+node.js展示
- 动态地展示在我们的前面页面上
本文主要分享对某个电商网站产生的用户访问日志(access.log)进行离线处理与分析的过程,基于MapReduce的处理方式,最后会统计出某一天不同省份访问该网站的uv与pv。
1 、数据源
在我们的场景中,Web应用的部署是如下的架构:
即比较典型的Nginx负载均衡+KeepAlive高可用集群架构
,在每台Web服务器上,都会产生用户的访问日志,业务需求方给出的日志格式如下:
1001 211.167.248.22 eecf0780-2578-4d77-a8d6-e2225e8b9169 40604 1 GET /top HTTP/1.0 408 null null 1523188122767
1003 222.68.207.11 eecf0780-2578-4d77-a8d6-e2225e8b9169 20202 1 GET /tologin HTTP/1.1 504 null Mozilla/5.0 (Windows; U; Windows NT 5.1)Gecko/20070309 Firefox/2.0.0.3 1523188123267
1001 61.53.137.50 c3966af9-8a43-4bda-b58c-c11525ca367b 0 1 GET /update/pass HTTP/1.0 302 null null 1523188123768
1000 221.195.40.145 1aa3b538-2f55-4cd7-9f46-6364fdd1e487 0 0 GET /user/add HTTP/1.1 200 null Mozilla/4.0 (compatible; MSIE 7.0; Windows NT5.2) 1523188124269
1000 121.11.87.171 8b0ea90a-77a5-4034-99ed-403c800263dd 20202 1 GET /top HTTP/1.0 408 null Mozilla/5.0 (Windows; U; Windows NT 5.1)Gecko/20070803 Firefox/1.5.0.12 1523188120263appid ip mid userid login_type request status http_referer user_agent time
其中:
appid包括 : web:1000,android:1001,ios:1002,ipad:1003
mid: 唯一的id此id第一次会种在浏览器的cookie里。如果存在则不再种。作为浏览器唯一标示。移动端或者pad直接取机器码。
login_type: 登录状态,0未登录、1:登录用户
request: 类似于此种 "GET /userList HTTP/1.1"
status: 请求的状态主要有:200 ok、404 not found、408 Request Timeout、500 Internal Server Error、504 Gateway Timeout等
http_referer:请求该url的上一个url地址。
user_agent: 浏览器的信息,例如:"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/47.0.2526.106 Safari/537.36"
time: 时间的long格式:1451451433818。
如果备份日志或者日志切割:
vim /opt/cut_nginx.sh
#!/bin/bash
#切割日志
datetime=$(date -d "-1 day" "+%Y%m%d")
log_path="/usr/local/nginx/logs"
pid_path="/usr/local/nginx/logs/nginx.pid"
[ -d $log_path/backup ] || mkdir -p $log_path/backup
if [ -f $pid_path ]
then
mv $log_path/access.log $log_path/backup/access.log-$datetime
kill -USR1 $(cat $pid_path)
find $log_path/backup -mtime +30 | xargs rm -f#mtime :文件被修改时间 atime:访问时间(文件中的数据库最后被访问的时间) ctime:改变时间(文件的元数据发生变化。比如权限,所有者等)
else
echo "Error,Nginx is not working!" | tee -a /var/log/messages
fi
chmod +x /opt/cut_nginx.sh
crontab -e 设置定时任务
0 0 * * * /opt/cut_nginx.sh
1、模拟生成实时数据
public class SimulateData {public static void main(String[] args) {BufferedWriter bw = null;try {bw = new BufferedWriter(new FileWriter("G:\\Scala\\实时统计每日的品类的点击次数\\data.txt"));int i = 0;while (i < 20000){long time = System.currentTimeMillis();int categoryid = new Random().nextInt(23);bw.write("ver=1&en=e_pv&pl=website&sdk=js&b_rst=1920*1080&u_ud=12GH4079-223E-4A57-AC60-C1A04D8F7A2F&l=zh-CN&u_sd=8E9559B3-DA35-44E1-AC98-85EB37D1F263&c_time="+time+"&p_url=http://list.iqiyi.com/www/"+categoryid+"/---.html");bw.newLine();i++;}} catch (IOException e) {e.printStackTrace();}finally {try {bw.close();} catch (IOException e) {e.printStackTrace();}}}
}
/*
ver=1&en=e_pv&pl=website&sdk=js&b_rst=1920*1080&u_ud=12GH4079-223E-4A57-AC60-C1A04D8F7A2F&l=zh-CN&u_sd=8E9559B3-DA35-44E1-AC98-85EB37D1F263&c_time=1526975174569&p_url=http://list.iqiyi.com/www/9/---.html
ver=1&en=e_pv&pl=website&sdk=js&b_rst=1920*1080&u_ud=12GH4079-223E-4A57-AC60-C1A04D8F7A2F&l=zh-CN&u_sd=8E9559B3-DA35-44E1-AC98-85EB37D1F263&c_time=1526975174570&p_url=http://list.iqiyi.com/www/4/---.html
ver=1&en=e_pv&pl=website&sdk=js&b_rst=1920*1080&u_ud=12GH4079-223E-4A57-AC60-C1A04D8F7A2F&l=zh-CN&u_sd=8E9559B3-DA35-44E1-AC98-85EB37D1F263&c_time=1526975174570&p_url=http://list.iqiyi.com/www/10/---.html*/
模拟数据实时的写入data.log:需要一直启动着:
#!/bin/bash
cat demo.csv | while read line
do
echo "$line" >> data.log
sleep 1
done
或者生成数据直接发送kafka
/*** 这里产生数据,就会发送给kafka,kafka那边启动消费者,就会接收到数据,这一步是用来测试生成数据和消费数据没有问题的,确定没问题后要关闭消费者,* 启动OnlineBBSUserLogss.java的类作为消费者,就会按pv,uv等方式处理这些数据。* 因为一个topic只能有一个消费者,所以启动程序前必须关闭kafka方式启动的消费者(我这里没有关闭关闭kafka方式启动的消费者也没正常啊) */
public class SparkStreamingDataManuallyProducerForKafkas extends Thread{//具体的论坛频道static String[] channelNames = new String[]{"Spark","Scala","Kafka","Flink","Hadoop","Storm","Hive","Impala","HBase","ML"};//用户的两种行为模式static String[] actionNames = new String[]{"View", "Register"};private static Producer<String, String> producerForKafka;private static String dateToday;private static Random random;//2、作为线程而言,要复写run方法,先写业务逻辑,再写控制@Overridepublic void run() {int counter = 0;//搞500条while(true){//模拟实际情况,不断循环,异步过程,不可能是同步过程counter++;String userLog = userlogs();System.out.println("product:"+userLog);//"test"为topicproducerForKafka.send(new KeyedMessage<String, String>("test", userLog));if(0 == counter%500){counter = 0;try {Thread.sleep(1000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}}private static String userlogs() {StringBuffer userLogBuffer = new StringBuffer("");int[] unregisteredUsers = new int[]{1, 2, 3, 4, 5, 6, 7, 8};long timestamp = new Date().getTime();Long userID = 0L;long pageID = 0L;//随机生成的用户ID if(unregisteredUsers[random.nextInt(8)] == 1) {userID = null;} else {userID = (long) random.nextInt((int) 2000);}//随机生成的页面IDpageID = random.nextInt((int) 2000); //随机生成ChannelString channel = channelNames[random.nextInt(10)];//随机生成action行为String action = actionNames[random.nextInt(2)];userLogBuffer.append(dateToday).append("\t").append(timestamp).append("\t").append(userID).append("\t").append(pageID).append("\t").append(channel).append("\t").append(action); //这里不要加\n换行符,因为kafka自己会换行,再append一个换行符,消费者那边就会处理不出数据return userLogBuffer.toString();}public static void main(String[] args) throws Exception {dateToday = new SimpleDateFormat("yyyy-MM-dd").format(new Date());random = new Random();Properties props = new Properties();props.put("zk.connect", "h71:2181,h72:2181,h73:2181");props.put("metadata.broker.list","h71:9092,h72:9092,h73:9092");props.put("serializer.class", "kafka.serializer.StringEncoder");ProducerConfig config = new ProducerConfig(props);producerForKafka = new Producer<String, String>(config);new SparkStreamingDataManuallyProducerForKafkas().start(); }
} /**
product:2017-06-20 1497948113827 633 1345 Hive View
product:2017-06-20 1497948113828 957 1381 Hadoop Register
product:2017-06-20 1497948113831 300 1781 Spark View
product:2017-06-20 1497948113832 1244 1076 Hadoop Register
**/
2、数据采集:获取原生数据
数据采集工作:使用Flume对于用户访问日志的采集,将采集的数据保存到HDFS中 (离线)、发送数据到kafka(实时)
2、flume发送数据到kafka
从data.log文件中读取实时数据到kafka:
第一步:配置Flume文件:(file2kafka.properties)
a1.sources = r1
a1.sinks = k1
a1.channels =c1a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data.loga1.channel.c1 = memorya1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = aura
a1.sinks.k1.brokerList = hodoop02:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 5a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
第四步:启动Flume命令
[hadoop@hadoop02 apache-flume-1.8.0-bin]$
bin/flume-ng agent --conf conf --conf-file /usr/local/flume/example/file2kafka.properties --name a1 -Dflume.root.logger=INFO,console
第三步:启动kafka消费者
[hadoop@hadoop03 kafka_2.11-1.0.0]$
bin/kafka-console-consumer.sh --zookeeper hadoop:2181 --from-beginning --topic aura
4 数据清洗:将不规整数据转化为规整数据(存入hdfs或者hive用于离线分析统计)
4.3.3 执行MapReduce程序
将上面的mr程序打包后上传到我们的Hadoop环境中,这里,对2018-04-08
这一天产生的日志数据进行清洗,执行如下命令:
yarn jar data-extract-clean-analysis-1.0-SNAPSHOT-jar-with-dependencies.jar\
cn.xpleaf.dataClean.mr.job.AccessLogCleanJob \
hdfs://ns1/input/data-clean/access/2018/04/08 \
hdfs://ns1/output/data-clean/access
5 数据处理:对规整数据进行统计分析
6、Kafka消费者,SparkStream时实计算商品销售额--redis
实战SparkStream+Kafka+Redis实时计算商品销售额_小哈-whzhaochao-CSDN博客_sparkstream+kafka
object OrderConsumer {//Redis配置val dbIndex = 0//每件商品总销售额val orderTotalKey = "app::order::total"//每件商品上一分钟销售额val oneMinTotalKey = "app::order::product"//总销售额val totalKey = "app::order::all"def main(args: Array[String]): Unit = {// 创建 StreamingContext 时间片为1秒val conf = new SparkConf().setMaster("local").setAppName("UserClickCountStat")val ssc = new StreamingContext(conf, Seconds(1))// Kafka 配置val topics = Set("order")val brokers = "127.0.0.1:9092"val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,"serializer.class" -> "kafka.serializer.StringEncoder")// 创建一个 direct streamval kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)//解析JSONval events = kafkaStream.flatMap(line => Some(JSON.parseObject(line._2)))// 按ID分组统计个数与价格总合val orders = events.map(x => (x.getString("id"), x.getLong("price"))).groupByKey().map(x => (x._1, x._2.size, x._2.reduceLeft(_ + _)))//输出orders.foreachRDD(x =>x.foreachPartition(partition =>partition.foreach(x => {println("id=" + x._1 + " count=" + x._2 + " price=" + x._3)//保存到Redis中val jedis = RedisClient.pool.getResourcejedis.select(dbIndex)//每个商品销售额累加jedis.hincrBy(orderTotalKey, x._1, x._3)//上一分钟第每个商品销售额jedis.hset(oneMinTotalKey, x._1.toString, x._3.toString)//总销售额累加jedis.incrBy(totalKey, x._3)RedisClient.pool.returnResource(jedis)})))ssc.start()ssc.awaitTermination()}
}
/*
id=4 count=3 price=7208
id=8 count=2 price=10152
id=7 count=1 price=6928
id=5 count=1 price=3327
id=6 count=3 price=20483
id=0 count=2 price=9882
*/
Redis客户端
object RedisClient extends Serializable {val redisHost = "127.0.0.1"val redisPort = 6379val redisTimeout = 30000lazy val pool = new JedisPool(new GenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)lazy val hook = new Thread {override def run = {println("Execute hook thread: " + this)pool.destroy()}}sys.addShutdownHook(hook.run)def main(args: Array[String]): Unit = {val dbIndex = 0val jedis = RedisClient.pool.getResourcejedis.select(dbIndex)jedis.set("test", "1")println(jedis.get("test"))RedisClient.pool.returnResource(jedis)}
}
DAU日活
// 转换处理val startuplogStream: DStream[Startuplog] = inputDstream.map {record =>val jsonStr: String = record.value()val startuplog: Startuplog = JSON.parseObject(jsonStr, classOf[Startuplog])val date = new Date(startuplog.ts)val dateStr: String = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(date)val dateArr: Array[String] = dateStr.split(" ")startuplog.logDate = dateArr(0)startuplog.logHour = dateArr(1).split(":")(0)startuplog.logHourMinute = dateArr(1)startuplog}// 利用redis进行去重过滤val filteredDstream: DStream[Startuplog] = startuplogStream.transform {rdd =>println("过滤前:" + rdd.count())//driver //周期性执行DataFrameval curdate: String = new SimpleDateFormat("yyyy-MM-dd").format(new Date)val jedis: Jedis = RedisUtil.getJedisClientval key = "dau:" + curdateval dauSet: util.Set[String] = jedis.smembers(key) //SMEMBERS key 获取集合里面的所有keyval dauBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dauSet)val filteredRDD: RDD[Startuplog] = rdd.filter {startuplog =>//executorval dauSet: util.Set[String] = dauBC.value!dauSet.contains(startuplog.mid)}println("过滤后:" + filteredRDD.count())filteredRDD}//去重思路;把相同的mid的数据分成一组 ,每组取第一个val groupbyMidDstream: DStream[(String, Iterable[Startuplog])] = filteredDstream.map(startuplog => (startuplog.mid, startuplog)).groupByKey()val distinctDstream: DStream[Startuplog] = groupbyMidDstream.flatMap {case (mid, startulogItr) =>startulogItr.take(1)}// 保存到redis中distinctDstream.foreachRDD { rdd =>// redis type set// key dau:2019-06-03 value : midsrdd.foreachPartition { startuplogItr =>//executorval jedis: Jedis = RedisUtil.getJedisClientval list: List[Startuplog] = startuplogItr.toListfor (startuplog <- list) {val key = "dau:" + startuplog.logDateval value = startuplog.midjedis.sadd(key, value)println(startuplog) //往es中保存}MyEsUtil.indexBulk(GmallConstant.ES_INDEX_DAU, list)jedis.close()}}
7、SparkStreaming接收kafka数据并处理--Hbase
创建一个HBase表:
实时统计每日的分类的点击次数,存储到HBase(HBase表示如何设计的,rowkey是怎样设计)
原文: SparkStreaming项目(实时统计每个品类被点击的次数)_potter-CSDN博客_sparkstreaming项目
object DauApp {def main(args: Array[String]): Unit = {val sparkConf: SparkConf = new SparkConf().setAppName("dau_app").setMaster("local[*]")val ssc = new StreamingContext(sparkConf, Seconds(5))val inputDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstant.KAFKA_TOPIC_STARTUP, ssc)// inputDstream.foreachRDD{rdd=>// println(rdd.map(_.value()).collect().mkString("\n"))// }// val dStream: DStream[String] = inputDstream.map { record =>// val jsonStr: String = record.value()// jsonStr// }//处理数据,完了,保存偏移量inputDstream.foreachRDD(rdd => {//手动指定分区的地方val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRangesprintln("长度=" + ranges.length)ranges.foreach(println)val result: RDD[(String, Int)] = rdd.map(_.value()).flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)result.foreach(println)result.foreachPartition(p => {val jedis: Jedis = RedisUtil.getJedisClientp.foreach(rdd2 => {// 数据数理逻辑jedis.hincrBy("wc1", rdd2._1, rdd2._2)})//把通过hset,把对应的partition和offset写入到redis中val map = new util.HashMap[String, String]()for (o <- ranges) {val offset = o.untilOffsetval partition = o.partitionval topic = o.topicval group_id = "gmall_consumer_group"map.put("offset", offset.toString)jedis.hmset("offsetKey", map)}jedis.close()})// 把偏移量的Array 写入到mysql中// ranges.foreach(rdd2 => {// // 思考,需要保存哪些数据呢? 起始的offset不需要 还需要加上 groupid// val pstm = conn.prepareStatement("replace into mysqloffset values (?,?,?,?)")// pstm.setString(1, rdd2.topic)// pstm.setInt(2, rdd2.partition)// pstm.setLong(3, rdd2.untilOffset)// pstm.setString(4, groupId)// pstm.execute()// pstm.close()// })})// 转换处理val startuplogStream: DStream[Startuplog] = inputDstream.map {record =>val jsonStr: String = record.value()val startuplog: Startuplog = JSON.parseObject(jsonStr, classOf[Startuplog])val date = new Date(startuplog.ts)val dateStr: String = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(date)val dateArr: Array[String] = dateStr.split(" ")startuplog.logDate = dateArr(0)startuplog.logHour = dateArr(1).split(":")(0)startuplog.logHourMinute = dateArr(1)startuplog}// 利用redis进行去重过滤val filteredDstream: DStream[Startuplog] = startuplogStream.transform {rdd =>println("过滤前:" + rdd.count())//driver //周期性执行DataFrameval curdate: String = new SimpleDateFormat("yyyy-MM-dd").format(new Date)val jedis: Jedis = RedisUtil.getJedisClientval key = "dau:" + curdateval dauSet: util.Set[String] = jedis.smembers(key) //SMEMBERS key 获取集合里面的所有keyval dauBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dauSet)val filteredRDD: RDD[Startuplog] = rdd.filter {startuplog =>//executorval dauSet: util.Set[String] = dauBC.value!dauSet.contains(startuplog.mid)}println("过滤后:" + filteredRDD.count())filteredRDD}//去重思路;把相同的mid的数据分成一组 ,每组取第一个val groupbyMidDstream: DStream[(String, Iterable[Startuplog])] = filteredDstream.map(startuplog => (startuplog.mid, startuplog)).groupByKey()val distinctDstream: DStream[Startuplog] = groupbyMidDstream.flatMap {case (mid, startulogItr) =>startulogItr.take(1)}// 保存到redis中distinctDstream.foreachRDD { rdd =>// redis type set// key dau:2019-06-03 value : midsrdd.foreachPartition { startuplogItr =>//executorval jedis: Jedis = RedisUtil.getJedisClientval list: List[Startuplog] = startuplogItr.toListfor (startuplog <- list) {val key = "dau:" + startuplog.logDateval value = startuplog.midjedis.sadd(key, value)println(startuplog) //往es中保存}MyEsUtil.indexBulk(GmallConstant.ES_INDEX_DAU, list)jedis.close()}}ssc.start()ssc.awaitTermination()}
}